02、体验Spark shell下RDD编程
1、Spark RDD介绍
RDD是Resilient Distributed Dataset，中文翻译是弹性分布式数据集。该类是Spark是核心类成员之一，是贯穿Spark编程的始终。初期阶段，我们可以把RDD看成是Java中的集合就可以了，在后面的章节中会详细讲解RDD的内部结构和工作原理。
2、Spark-shell下实现对本地文件的单词统计
2.1思路
word count是大数据学习的经典案例，很多功能实现都可以归结为是word count的使用。工作过程为使用SparkContext对象的textFile方法加载文件形成Spark RDD1，RDD1中每个元素就是文件中的每一行文本，然后对RDD的每个元素进行压扁flatMap操作，形成RDD2，RDD2中每个元素是将RDD1的每行拆分出来产生的单词，因此RDD2就是单词的集合，然后再对RDD2进行标一成对，形成（单词,1）的元组的集合RDD3，最后对RDD3进行按照key进行聚合操作形成RDD4，最终将RDD4计算后得到的集合就是每个单词的数量
2.2 处理流程
App->SparkContext: textFile加载文件
SparkContext->RDD1: 创建RDD
RDD1-->App: 返回RDD1
App->RDD1: flatMap压扁操作
RDD1->RDD2: 产生RDD2
RDD2-->App: 返回RDD2
App->RDD2: map标一成对
RDD2->RDD3: 产生RDD3
RDD3-->App: 返回RDD3
App->RDD3: reduceByKey聚合
RDD3->RDD4: 产生RDD4
RDD4-->App: 返回RDD4
App->RDD4: collect收集结果数据
2.3 分步实现代码
// 进入spark shell环境
$>spark-shell
// 1.加载文件
scala>val rdd1 = sc.textFile("file:///homec/centos/1.txt")
// 2.压扁每行
scala>val rdd2 = rdd1.flatMap(_.split(" "))
// 3.标1成对
scala>val rdd3 = rdd2.map(w=>(w,1))
// 4.按照key聚合每个key下的所有值
scala>val rdd4 = rdd3.reduceByKey(_+_)
// 5.显式数据
scala>rdd4.collect()
2.4 一步实现代码
$scala>sc.textFile("file:///home/centos/1.txt")
.flatMap(_.split(" "))
.map((_,1))
.reduceByKey(_+_)
.collect
3、Spark-shell下实现对气温数据的最大最小聚合
3.1 思路分析
气温数据数各年度内气温列表，将每一行变换成（year,temp）元组后，按照yearn进行聚合即可。
3.2 处理流程
App->SparkContext: textFile加载文件
SparkContext->RDD1: 产生RDD1
RDD1-->App: 返回RDD1
App->RDD1: map变换每行为(year,(max,min))元组
RDD1->RDD2: 产生RDD2
RDD2-->App: 返回RDD2
App->RDD2: reduceByKey双聚合气温极值
RDD2->RDD3:产生RDD3
App->RDD3: collect()收集结果
3.3 分步实现代码
// 进入spark shell环境
$>spark-shell
// 1.加载气温数据文件
scala>val rdd1 = sc.textFile("/home/centos/temps.dat")
// 2.压扁每行
scala>val rdd2 = rdd1.flatMap(e=>{
val arr = e.split(" ")
(arr(0).toInt, (arr(1).toInt ,arr(1).toInt))
})
// 3.reduceByKey
scala>val rdd3 = rdd2.reduceByKey((a,b)=>{
import scala.math
(math.max(a(0),b(0)) , math.min(a(1),b(1)))
})
// 4.收集日志
scala>rdd3.collect()
3.4 一步实现代码
$scala>sc.textFile("file:///home/centos/temps.dat")
.map(line=>{
val arr = line.split(" ")
(arr(0).toInt,(arr(1).toInt , arr(1).toInt))
})
.reduceByKey((a,b)=>{
import scala.math
(math.max(a(0) , b(0)) , math.min(a(1) , b(1)))
})
.collect()