一.RDD
28.RDD_为什么需要RDD
29.RDD_定义
30.RDD_五大特性总述
31.RDD_五大特性1
32.RDD_五大特性2
33.RDD_五大特性3
34.RDD_五大特性4
35.RDD_五大特性5
36.RDD_五大特性总结
37.RDD_创建概述
38.RDD_并行化创建
演示代码：
// 获取当前 RDD 的分区数
@Since ( "1.6.0" )
final def getNumPartitions : Int =
partitions . length
// 显示出 RDD 被分配到不同分区的信息
/**Return an RDD created by coalescing all
elements within each partition into an
array.*/
def glom (): RDD [ Array [ T ]]
1
2
3
4
5
6
package com . itbaizhan . rdd
//1. 导入 SparkConf 类、 SparkContext
SparkContext }
object CreateByParallelize {
def main ( args : Array [ String ]): Unit = {
//2. 构建 SparkConf 对象。并设置本地运行和程序的
名称
val conf = new
SparkConf (). setMaster ( "local[2]" ). setAppName
( "CreateRdd1" )
//3. 构建 SparkContext 对象
val sc = new SparkContext ( conf )
//4. 通过并行化创建 RDD 对象：将本地集合 -> 分布式的
RDD 对象
1
2
3
4
5
6
7
8
9
10
11
12
79
//val rdd: RDD[Int] =
sc.parallelize[Int](List(1, 2, 3, 4, 5, 6,
7, 8))
val rdd : RDD [ Int ] =
sc . parallelize ( List ( 1 , 2 , 3 , 4 , 5 , 6 , 7 ,
8 ), 3 )
//5. 输出默认的分区数
//5.1
setMaster("local[*]")&¶llelize(List(1,
2, 3, 4, 5, 6, 7, 8))
//println(" 默认分区
数： "+rdd.getNumPartitions)//8, 默认当前系统的
CPU 数
//5.2
setMaster("local[2]")&¶llelize(List(1,
2, 3, 4, 5, 6, 7, 8))
//println(" 默认分区
数： "+rdd.getNumPartitions)//2
//5.3
setMaster("local[2]")&¶llelize(List(1,
2, 3, 4, 5, 6, 7, 8),3)
println ( " 默认分区
数： " + rdd . getNumPartitions ) //3
//6.collect 方法：将 rdd 对象中每个分区的数据，都
发送到 Driver ，形成一个 Array 对象
val array1 : Array [ Int ] = rdd . collect ()
println ( "rdd.collect()=" + array1 . mkString ( ",
" ))
//7. 显示出 rdd 对象中元素被分布到不同分区的数据信
息
13
14
15
16
17
18
19
20
21
22
23
24
25
80 运行结果：
实时效果反馈
1. 以下关于并行化创建 RDD 的描述错误的是：
A
通过并行化集合创建，将本地集合对象转分布式 RDD 。
B
parallelize() 方法必须传递两个参数。
C
parallelize 没有给定分区数 , 默认分区数等于执行程序的当前
服务器 CPU 核数。
答案：
val array2 : Array [ Array [ Int ]] =
rdd . glom (). collect ()
println ( "rdd.glom().collect() 的内容是 :" )
/*for(eleArr<- array2){
println(eleArr.mkString(","))
}*/
array2 . foreach ( eleArr => println ( eleArr . mkStr
ing ( "," )))
}
}
26
27
28
29
30
31
32
33
默认分区数： 3
rdd.collect()=1,2,3,4,5,6,7,8
rdd.glom().collect() 的内容是 :
1,2
3,4,5
6,7,8
39.RDD_读取文件创建RDD
40.RDD_读取小文件创建RDD
扩展 wholeTextFiles 适合读取一堆小文件：
//path 指定小文件的路径目录
//minPartitions 最小分区数 可选参数
def wholeTextFiles ( path :
String , minPartitions : Int =
defaultMinPartitions ): RDD [( String , String )]
1
2
3
85 代码演示：
package com . itbaizhan . rdd
//1. 导入类
SparkContext }
object CreateByWholeTextFiles {
def main ( args : Array [ String ]): Unit = {
//2. 构建 SparkConf 对象，并设置本地运行和程序名
称
val conf : SparkConf = new
SparkConf (). setMaster ( "local[*]" ). setAppName
( "WholeTextFiles" )
//3. 使用 conf 对象构建 SparkContet 对象
val sc = new SparkContext ( conf )
//5. 读取指定目录下的小文件
val rdd : RDD [( String , String )] =
sc . wholeTextFiles ( "data/tiny_files" )
//(filePath1, " 内容 1"),(filePath2, " 内容
2"),...,(filePathN, " 内容 N")
val tuples : Array [( String , String )] =
rdd . collect ()
tuples . foreach ( ele => println ( ele . _1 , ele . _2 ))
//6. 获取小文件中的内容
val array : Array [ String ] =
rdd . map ( _ . _2 ). collect ()
println ( "---------------------------" )
println ( array . mkString ( "|" ))
//4. 关闭 sc 对象
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
86 运行输出结果 :
RDD_ 算子概述
定义： 分布式集合 RDD 对象的方法被称为算子
算子分类：
Transformation 转换算子
1
Action 行动算子
2
sc . stop ()
}
}
22
23
24
(file:/D:/codes/itbaizhan/sparkdemo/data/tin
y_files/file1.txt,hello Linux
hello Zookeper
hello Maven
hello hive
hello spark)
(file:/D:/codes/itbaizhan/sparkdemo/data/tin
y_files/file2.txt,Spark Core
Spark RDD
Spark Sql)
----------------
hello Linux
hello Zookeper
hello Maven
hello hive
hello spark|Spark Core
Spark RDD
Spark Sql
41.RDD_算子概述
42.RDD_转换算子map
43.RDD_转换算子flatmap
44.RDD_转换算子reducebykey
45.RDD_转换算子filter
46.RDD_转换算子distinct
47.RDD_转换算子glom
48.RDD_转换算子groupby
object RddGroupBy {
def main ( args : Array [ String ]): Unit = {
//2. 构建 SparkConf 对象，并设置本地运行和程序名
称
val conf : SparkConf = new
SparkConf (). setMaster ( "local[*]" ). setAppName
( "groupBy" )
//3. 使用 conf 对象构建 SparkContet 对象
val sc = new SparkContext ( conf )
//5. 创建 Rdd
val rdd : RDD [( Char , Int )] =
sc . parallelize ( Array (( 'a' , 1 ), ( 'a' , 2 ),
( 'b' , 1 ), ( 'b' , 2 ), ( 'a' , 3 ), ( 'a' , 4 )))
//6. 通过 groupBy 算子对 rdd 对象中的数据进行分组
//groupBy 插入的函数的用意是指定按照谁进行分组
// 分组后的结果是有二元组组成的 RDD
val gbRdd : RDD [( Char , Iterable [( Char ,
Int )])] = rdd . groupBy ( tupEle => tupEle . _1 )
// 收集到 Driver 端
val result1 : Array [( Char ,
Iterable [( Char , Int )])] = gbRdd . collect ()
//(a,CompactBuffer((a,1), (a,2), (a,3),
(a,4))),(b,CompactBuffer((b,1), (b,2)))
println ( result1 . mkString ( "," ))
//7. 使用 map 转换算子
//(a,List((a,1), (a,2), (a,3), (a,4))),
(b,List((b,1), (b,2)))
val result2 : Array [( Char , List [( Char ,
Int )])] = gbRdd . map ( tup => ( tup . _1 ,
tup . _2 . toList )). collect ()
println ( result2 . mkString ( "," ))
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
104 实时效果反馈
1. 以下关于
rdd.groupBy(tupEle => tupEle._1)
的描述错误的是：
A
groupBy 传入的函数的意思是 : 通过这个函数 , 确定按照谁来
分组。
B
groupBy 方法适用于元素为元祖类型的 RDD ，元祖元素的个
数只能为 2 。
C
groupBy 方法适用于元素为元祖类型的 RDD ，元祖元素的个
数 >=2 。
答案：
1=>B
49.RDD_转换算子groupbyKey
50.RDD_转换算子sortby
51.RDD_转换算子sortbyKey
52.RDD_转换算子union并集
53.RDD_转换算子交集和差集
54.RDD_转换算子关联算子
55.RDD_转换算子partitionBy
56.RDD_转换算子mapPatitions
57.RDD_转换算子sample
58.RDD_行动算子foreachPartition
59.RDD_行动算子foreach
60.RDD_行动算子saveAsTestFile
61.RDD_行动算子countByKey
62.RDD_行动算子reduce
63.RDD_行动算子fold
64.RDD_行动算子first_take_count
65.RDD_行动算子top_takeOrderd
66.RDD_行动算子takeSample
二.内核进阶
67.内核进阶_DAG概述
68.内核进阶_血缘关系
69.内核进阶_宽窄依赖关系
70.内核进阶_stage划分
71.内核进阶_任务调度概述
72.内核进阶_管道计算模式上
73.内核进阶_管道计算模式下
74.内核进阶_cache缓存
75.内核进阶_checkpoint检查点
76.内核进阶_cache和checkpoint区别
77.内核进阶_并行度
78.内核进阶_广播变量
79.内核进阶_累加器一
80.内核进阶_累加器二
81.内核进阶_累加器之重复计算
82.内核进阶_项目实战PVUV需求分析
83.内核进阶_项目实战PV分析
84.内核进阶_项目实战UV分析
85.内核进阶_二次排序实战
86.内核进阶_分组取topN实战
87.内核进阶_卡口统计项目需求分析
88.内核进阶_卡口统计项目统计正常的卡口
89.内核进阶_卡口统计项目TOP5
90.内核进阶_卡口统计项目统计不同区域同时出现的车辆
91.内核进阶_卡口统计项目统计某卡口下通过的车辆轨迹一
92.内核进阶_卡口统计项目统计某卡口下通过的车辆轨迹二
93.内核进阶_卡口统计项目统计某卡口下通过的车辆轨迹三
94.内核进阶_卡口统计项目统计某卡口下通过的车辆轨迹四
