2. PySpark——RDD编程入门
gitee仓库：gitee仓库
觉得有用的话，点个赞，点个收藏呗
给人点赞，手留余香
2.1 程序执行入口SparkContext对象
Spark RDD 编程的程序入口对象是SparkContext对象(不论何种编程语言)
只有构建出SparkContext, 基于它才能执行后续的API调用和计算
本质上, SparkContext对编程来说, 主要功能就是创建第一个RDD出来
代码演示：
# coding:utf8
# 导入Spark相关包
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
# 构建SparkConf对象
conf = SparkConf().setAppName ("helloSpark").setMaster("local[*]")
# 构建SparkContext执行环境入口对象
sc = SparkContext(conf=conf)
master的种类：
- local：local[N]:表示以N核CPU执行，local[*]:给予local进程 所有CPU核心的使用权
- standlone：spark：//node1:7077
- yarn 模式
2.2 RDD的创建
RDD的创建主要有2种方式:
• 通过并行化集合创建 ( 本地对象 转 分布式RDD )
• 读取外部数据源 ( 读取文件 )
2.2.1 并行化创建
概念：并行化创建，是指将本地集合转向分布式RDD，这一步就是分布式的开端：本地转分布式
API：
rdd = spakcontext.parallelize(参数1，参数2)
参数1 集合对象即可，比如list
参数2 分区数
完整代码：
# coding:utf8
from pyspark import SparkConf,SparkContext
if __name__ == '__main__':
# 0. 构建Spark执行环境
conf = SparkConf().setAppName("create rdd").setMaster("local[*]")
sc = SparkContext(conf=conf)
# sc 对象的parallelize方法，可以将本地集合转换成RDD返回给你
data = [1,2,3,4,5,6,7,8,9]
rdd = sc.parallelize(data,numSlices=3)
print(rdd.collect())
执行结果：
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
[1, 2, 3, 4, 5, 6, 7, 8, 9]
Process finished with exit code 0
2.2.2 获取RDD分区数
getNumPartitions API :获取RDD分区数量，返回值是Int数字
用法：rdd.getNumPartitions()
例如，基于上述代码设置了3为分区数，调用以下代码
print(rdd.getNumPartitions())
则会输出结果：3
完整案例代码：01_create_parallelize.py
# coding:utf8
# 导入Spark相关包
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
# 0. 初始化执行环境 构建SparkContext对象,本地集合--> 分布式对象（RDD）
conf = SparkConf().setAppName ("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
# 演示通过并行化集合的方式去创建RDD
rdd = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9])
# parallelize方法，没有给定分区数，默认分区数是多少？ 根据CPU核心来定
print("默认分区数：", rdd.getNumPartitions())
rdd = sc.parallelize([1, 2, 3], 3)
print("分区数：", rdd.getNumPartitions())
# collect方法，是将RDD（分布式对象）中每个分区的数据，都发送到Driver中，形成一个Python List对象
# collect：分布式 转--> 本地集合
print("rdd的内容是：", rdd.collect())
print(type(rdd.collect()))
输出结果：
默认分区数： 8
分区数： 3
rdd的内容是： [1, 2, 3]
<class 'list'>
2.2.3 读取文件创建
textFileAPI
这个API可以读取本地数据，也可以读取hdfs数据
使用方法 ：
sparkcontext.textFile(参数1,参数2)
参数1，必填，文件路径 支持本地文件 支持HDFS 也支持一些比如S3协议
参数2 可选，表示最小分区数量
注意：参数2 话语权不足，spark有自己的判断，在它允许的范围内，参数2有效果，超出spark允许的范围，参数2失效
案例代码：02_create_textFile.py
# coding : utf8
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("02_create_textFile").setMaster("local[*]")
sc = SparkContext(conf=conf)
# 通过textFile API 读取数据
# 读取本地文件数据
file_rdd1 = sc.textFile("../data/input/words.txt")
print("默认读取分区数：", file_rdd1.getNumPartitions())
print("file_rdd1 内容：", file_rdd1.collect())
#
# # 加最小分区数的测试
file_rdd2 = sc.textFile("../data/input/words.txt",3)
file_rdd3 = sc.textFile("../data/input/words.txt",100)
print("file_rdd2 分区数：", file_rdd2.getNumPartitions())
print("file_rdd3 分区数：", file_rdd3.getNumPartitions())
# 读取hdfs文件数据测试
hdfs_rdd = sc.textFile("hdfs://Tnode1:8020/input/words.txt")
print("hdfs_rdd 分区数：", hdfs_rdd.getNumPartitions())
print("hdfs_rdd 内容：", hdfs_rdd.collect())
输出结果：
默认读取分区数： 2
file_rdd1 内容： ['hello spark', 'hello hadoop', 'hello flink']
file_rdd2 内容： 4
file_rdd3 内容： 38
hdfs_rdd 分区： 2
hdfs_rdd 内容： ['hello spark', 'hello hadoop', 'hello flink']
wholeTextFile 读取文件的API，有个适用场景：适合读取一堆小文件
这个API是小文件读取专用
用法：
sparkcontext.textFile(参数1,参数2)
# 参数1，必填，文件路径 支持本地文件 支持HDFS 也支持一些比如S3协议
# 参数2 可选，表示最小分区数量
# 注意：参数2 话语权不足，spark有自己的判断，在它允许的范围内，参数2有效果，超出spark允许的范围，参数2失效
这个API偏向于少量分区读取数据
因为，这个API表明了自己是小文件读取专用，那么文件的数据很小、分区很多，
导致shuffle的几率更高，所以尽量少分区读取数据
案例代码：03_create_wholeTextFile.py
# coding:utf8
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
# 读取小文件文件夹
rdd = sc.wholeTextFiles("../data/input/tiny_files")
print(rdd.collect())
print(rdd.map(lambda x: x[1]).collect())
输出结果：
[('file:/tmp/pycharm_project_937/PySpark01/data/input/tiny_files/1.txt', 'hello spark\r\nhello hadoop\r\nhello flink'), ('file:/tmp/pycharm_project_937/PySpark01/data/input/tiny_files/2.txt', 'hello spark\r\nhello hadoop\r\nhello flink'), ('file:/tmp/pycharm_project_937/PySpark01/data/input/tiny_files/3.txt', 'hello spark\r\nhello hadoop\r\nhello flink'), ('file:/tmp/pycharm_project_937/PySpark01/data/input/tiny_files/4.txt', 'hello spark\r\nhello hadoop\r\nhello flink'), ('file:/tmp/pycharm_project_937/PySpark01/data/input/tiny_files/5.txt', 'hello spark\r\nhello hadoop\r\nhello flink')]
['hello spark\r\nhello hadoop\r\nhello flink', 'hello spark\r\nhello hadoop\r\nhello flink', 'hello spark\r\nhello hadoop\r\nhello flink', 'hello spark\r\nhello hadoop\r\nhello flink', 'hello spark\r\nhello hadoop\r\nhello flink']
2.3 RDD算子
算子是什么？
算子：分布式集合对象上的API称之为算子
方法、函数：本地对象的API，叫做方法、函数
算子：分布式对象的API，叫做算子
算子分类
RDD的算子 分成2类
- Transformation：转换算子
- Action：动作（行动）算子
Transformation 算子：
定义：RDD的算子，返回值任然是一个RDD的，称之为转换算子
特性：这类算子lazy 懒加载的，如果没有action算子，Transformation算子是不工作的
Action算子
定义：返回值不是rdd的就是action算子
对于这两类算子来说，
Transformation算子，相当于在构建执行计划，action是一个指令让这个执行计划开始工作。如果没有
action，Transformation算子之间的迭代关系，就是一个没有通电的流水线，只有
action到来，这个数据处理的流水线才开始工作
2.4 常用Transformation算子
2.4.1 map算子
演示代码：04_operators_map.py
# coding:utf8
from pyspark import SparkConf, SparkContext
def addNum(data):
return data * 10
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd1 = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9], 4)
rdd2 = rdd1.map(lambda x: x * 10)
rdd3 = rdd1.map(addNum)
result = rdd2.collect()
print(result)
print(rdd3.collect())
输出结果：
[10, 20, 30, 40, 50, 60, 70, 80, 90]
[10, 20, 30, 40, 50, 60, 70, 80, 90]
对于传入参数的lambda表达式
传入方法作为传参的时候，可以选择
- 定义方法，传入其方法名
- 使用lambda 匿名方法的方式
一般，如果方法体可以一行写完，用lambda方便。
如果方法体复杂，就直接定义方法更方便
2.4.2 flatMap算子
功能：对rdd执行map操作，然后进行解除嵌套操作
解除嵌套：
演示代码：05_operators_flatMap.py
# coding:utf8
from pyspark import SparkConf, SparkContext
def addNum(data):
return data * 10
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd1 = sc.parallelize(["hadoop spark hadoop", "spark hadoop hadoop", "hadoop flink spark"])
# 得到所有的单词，组成rdd,flatMap的传入参数和map一致，就是给map逻辑用的，解除嵌套无需逻辑（传参）
rdd2 = rdd1.flatMap(lambda line: line.split(" "))
print(rdd2.collect())
输出结果：
['hadoop', 'spark', 'hadoop', 'spark', 'hadoop', 'hadoop', 'hadoop', 'flink', 'spark']
注意：flatMap只适合用于有“嵌套”的rdd，直接用于没有嵌套的rdd会报错
2.4.3 reduceByKey算子
功能：针对KV型的RDD，自动按照key分组，然后根据你提供的聚合逻辑，完成组内数据（value）的聚合操作。
用法：
rdd.reduceByKey(func)
# func:(V,V) ——>V
# 接收2个传入参数（类型要一致），返回一个返回值，类型和传入要求一致。
reduceByKey的聚合逻辑是：
比如，有[1,2,3,4,5],然后聚合函数是：lambda a,b: a+ b
注意：reduceByKey中接收的函数，只负责聚合，不理会分组
分组是自动
byKey来分组的。
代码演示：06_operators_reduceByKey.py
# coding:utf8
from pyspark import SparkConf, SparkContext
def addNum(data):
return data * 10
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([('a', 1), ('a', 1), ('b', 1), ('b', 1), ('a', 1)])
rdd2 = sc.parallelize([('a', 1), ('a', 11), ('b', 3), ('b', 1), ('a', 5)])
rdd3 = sc.parallelize([('a', 1), ('a', 11), ('b', 3), ('b', 1), ('a', 5)])
rdd = rdd.reduceByKey(lambda a, b: a + b)
rdd2 = rdd2.map(lambda x: (x[0], x[1] * 10))
# 只操作value的算子
rdd3 = rdd3.mapValues(lambda value: value * 10)
# recudeByKey 对相同key的数据执行聚合相加
print(rdd.collect())
print(rdd2.collect())
print(rdd3.collect())
输出结果：
[('a', 3), ('b', 2)]
[('a', 10), ('a', 110), ('b', 30), ('b', 10), ('a', 50)]
[('a', 10), ('a', 110), ('b', 30), ('b', 10), ('a', 50)]
2.4.4 WordCount回顾
代码演示：07_wordcount_example.py
# coding:utf8
from pyspark import SparkContext, SparkConf
if __name__ == '__main__':
# 构建SparkConf对象
conf = SparkConf().setAppName("test").setMaster("local[*]")
# 构建SparkContext执行环境入口对象
sc = SparkContext(conf=conf)
# 1.读取文件获取数据 构建RDD
file_rdd = sc.textFile(r"../data/input/words.txt")
# 2. 通过flatMap API取出所有的单词
word_rdd = file_rdd.flatMap(lambda x: x.split(" "))
# 3.将单词转换成元组，key是单词，value是1
word_with_one_rdd = word_rdd.map(lambda word:(word,1))
# 4. 用reduceByKey 对单词进行分组并进行value的聚合
result_rdd = word_with_one_rdd.reduceByKey(lambda a,b:a+b)
# 5. 通过collect算子，将rdd的数据收集到Driver中，打印输出
print(result_rdd.collect())
输出结果：
[('hadoop', 1), ('hello', 3), ('spark', 1), ('flink', 1)]
2.4.5 groupBy算子
功能：将rdd的数据进行分组
语法：
rdd.groupBy(func)
# func 函数
# func:(T)——>k
# 函数要求传入一个参数，返回一个返回值，类型无所谓
# 这个函数是 拿到你返回值后，将所有相同返回值的放入一个组中
# 分组完成后，每一个组是一个二元元组，key就是返回值，所有同组的数据放入一个迭代器对象中作为value
代码演示：08_oprators_groupBy.py
# coding:utf8
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([('a', 1), ('a', 1), ('b', 1), ('b', 1), ('b', 1)])
# 通过groupBy对数据进行分组
# groupBy传入的函数的意思是：通过这个函数，确定按照谁来分组(返回谁即可)
# 分组规则和SQL是一致的，也就是相同的在一个组（Hash分组）
result = rdd.groupBy(lambda t: t[0])
print(result.collect())
print("hello")
print(result.map(lambda t: (t[0], list(t[1]))).collect())
输出结果：
[('a', <pyspark.resultiterable.ResultIterable object at 0x7f85fa80eca0>), ('b', <pyspark.resultiterable.ResultIterable object at 0x7f85fa80ebb0>)]
hello
[('a', [('a', 1), ('a', 1)]), ('b', [('b', 1), ('b', 1), ('b', 1)])]
2.4.6 Filter算子
功能：过滤，把想要的数据进行保留
语法：
rdd.filter(func)
# func:(T)——>bool 传入1个随意类型参数进来，返回值必须是True or False
返回值是True的数据被保留，False的数据被丢弃
代码演示：09_operators_filter.py
# coding:utf8
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([1, 2, 3, 4, 5, 6])
# 通过Filter算子，过滤奇数,filter 只返回true的值
result = rdd.filter(lambda x: x % 2 == 1)
print(result.collect())
输出结果：
[1, 3, 5]
2.4.7 distinct算子
功能：对RDD数据进行去重，返回新的RDD
语法：
rdd.distinct(参数1)
# 参数1，去重分区数量，一般不用传
演示代码：10_operators_distinct.py
# coding:utf8
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([1, 1, 1, 2, 2, 2, 3, 3, 3])
# distinct 进行RDD数据去重操作
print(rdd.distinct().collect())
rdd2 = sc.parallelize([('a', 1), ('a', 1), ('a', 3)])
print(rdd2.distinct().collect())
输出结果：
[1, 2, 3]
[('a', 3), ('a', 1)]
2.4.8 union算子
功能：2个rdd合并成1个rdd返回
用法：rdd.union(other_rdd)
注意：只合并，不会去重
代码演示：11_operators_union.py
# coding:utf8
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd1 = sc.parallelize([1, 1, 3, 3])
rdd2 = sc.parallelize(["a","b","a"])
rdd3 = rdd1.union(rdd2)
print(rdd3.collect())
print(rdd3.distinct().collect())
"""
1. 可以看到union算子是不会去重的
2. RDD的类型不同也是可以合并的
"""
输出结果：
[1, 1, 3, 3, 'a', 'b', 'a']
[1, 3, 'b', 'a']
2.4.9 join算子
功能：对两个RDD执行JOIN操作（可实现SQL的内、外连接）
注意：join算子只能用于二元元组
语法：
rdd.join(other_rdd) #内连接
rdd.leftOuterJoin(other_rdd) # 左外
rdd.rightOuterJoin(other_rdd) # 右外
代码演示：12_operators_join.py
# coding:utf8
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd1 = sc.parallelize([(1001, "张三"), (1002, '李四'), (1003, '王五'), (1004, '赵六')])
rdd2 = sc.parallelize([(1001, "销售部"), (1002, '科技部')])
# 通过join算子来进行rdd之间的关联
# 对于join算子来说 关联条件 按照二元元组的key来进行关联
# 内连接
print(rdd1.join(rdd2).collect())
# 左外连接
print(rdd1.leftOuterJoin(rdd2).collect())
# 右外连接
print(rdd1.rightOuterJoin(rdd2).collect())
输出结果：
[(1001, ('张三', '销售部')), (1002, ('李四', '科技部'))]
[(1001, ('张三', '销售部')), (1002, ('李四', '科技部')), (1003, ('王五', None)), (1004, ('赵六', None))]
[(1001, ('张三', '销售部')), (1002, ('李四', '科技部'))]
2.4.10 intersection 算子
功能：求2个rdd的交集，返回一个新rdd
用法：rdd.intersection(other_rdd)
代码演示：13_operators_intersection.py
# coding:utf8
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd1 = sc.parallelize([('a',1),('a',3)])
rdd2 = sc.parallelize([('a',1),('b',3)])
# 通过intersection算子求RDD之间的交集，将交集取出，返回新RDD
rdd3 = rdd1.intersection(rdd2)
print(rdd3.collect())
输出结果：
[('a', 1)]
2.4.11 glom算子
功能：将RDD的数据，加上嵌套，这个嵌套按照分区来进行
比如RDD数据[1,2,3,4,5]有两个分区
那么，被glom后，数据变成：[[1,2,3],[4,5]]
使用方法：rdd.glom()
代码演示：14_operators_glom.py
# coding:utf8
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10],2)
rdd2 = sc.parallelize([1,2,3,4,5,6,7,8,9,10])
print(rdd.glom().collect())
print(rdd.glom().flatMap(lambda x:x).collect()) # 用flatMap解嵌套
print(rdd2.glom().collect())
输出结果：
[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
[[1], [2], [3], [4, 5], [6], [7], [8], [9, 10]]
2.4.12 groupByKey算子
功能：针对KV型RDD，自动按照key分组
用法：rdd.groupByKey() 自动按照key分组
代码演示：15_operators_groupByKey.py
# coding:utf8
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([('a', 1), ('a', 1), ('a', 1,), ('b', 1), ('b', 1)])
rdd2 = rdd.groupByKey()
print(rdd2.map(lambda x:(x[0],list(x[1]))).collect())
输出结果：
[('a', [1, 1, 1]), ('b', [1, 1])]
2.4.13 sortBy算子
功能：对RDD数据进行排序，基于你指定的排序依据
语法：
rdd.sortBy(func,ascending=False,numPartitions=1)
# func:(T)——>U:告知按照rdd中的哪个数据进行排序，比如lambda x:x[1] 表示按照rdd中的第二列元素进行排序
# ascending = True升序；False 降序
# numPartition：用多少分区来排序
注意：如果要全局有序，排序分区数请设置为1，因为生产环境下，分区数大于1，很可能只得到局部有序的结果
代码演示：16_operators_sortBy.py
# coding:utf8
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([('g', 3), ('c', 1), ('b', 2,), ('a', 9), ('h', 10), ('i', 4), ('l', 26,), ('o', 1), ('d', 7)])
# 使用sortBy对rdd进行排序
# 参数1 函数，表示的是，告诉spark按照数据的哪个列进行排序
# 参数2 bool，True表示升序，False表示降序
# 参数3 分区数设置
"""注意：如果要全局有序，排序分区数请设置为1，因为生产环境下，分区数大于1，很可能只得到局部有序的结果"""
rdd2 = rdd.sortBy(lambda x:x[1],ascending=True,numPartitions=3)
rdd3 = rdd.sortBy(lambda x:x[0],ascending=True,numPartitions=8)
print(rdd2.collect())
print(rdd3.collect())
输出结果：
[('c', 1), ('o', 1), ('b', 2), ('g', 3), ('i', 4), ('d', 7), ('a', 9), ('h', 10), ('l', 26)]
[('a', 9), ('b', 2), ('c', 1), ('d', 7), ('g', 3), ('h', 10), ('i', 4), ('l', 26), ('o', 1)]
2.4.14 sortByKey
功能：针对KV型RDD，按照key进行排序
语法：
sortByKey(ascending=True,numPartitions=None,keyfunc=<function RDD,<lambda>>)
- ascending:升序或降序，True升序，False降序，默认是升序
- numPartitions：按照几个分区进行排序，如果全局有序，设置为1
- keyfunc：在排序前对key进行处理，语法是：(k)——>U,一个参数传入，返回一个值
代码演示：17_operators_sortByKey.py
# coding:utf8
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([('g', 3), ('A', 1), ('B', 2,), ('A', 9), ('h', 10), ('i', 4), ('l', 26,), ('o', 1), ('d', 7)])
# 调用了忽略大小写的函数
print(rdd.sortByKey(ascending=True, numPartitions=1, keyfunc=lambda key: str(key).lower()).collect())
输出结果：
[('A', 1), ('A', 9), ('B', 2), ('d', 7), ('g', 3), ('h', 10), ('i', 4), ('l', 26), ('o', 1)]
2.4.15 综合案例
代码演示：18_operators_demo.py
# coding:utf8
import json
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
# 读取数据文件
file_rdd = sc.textFile("../data/input/order.text")
# 进行rdd数据的split 按照|符号进行，得到一个json数据
jsons_rdd = file_rdd.flatMap(lambda line: line.split("|"))
# 通过python内置的json库，完成json字符串到字典对象的转换
dict_rdd = jsons_rdd.map(lambda json_str: json.loads(json_str))
# 过滤数据，只保留北京的数据
beijing_rdd = dict_rdd.filter(lambda d: d['areaName'] == '北京')
# 组合北京和商品类型形成的字符串
category_rdd = beijing_rdd.map(lambda x: x['areaName'] + '_' + x['category'])
# 对结果集进行去重操作
result_rdd = category_rdd.distinct()
# 输出
print(result_rdd.collect())
输出结果：
['北京_平板电脑', '北京_家具', '北京_书籍', '北京_食品', '北京_服饰', '北京_手机', '北京_家电', '北京_电脑']
2.4.16 将案例提交到yarn运行
改动1：加入环境变量，让pycharm运行yarn的时候，知道hadoop的配置在哪，可以去读取yarn的信息
import os
from defs_19 import city_with_category
# 导入自己写的函数时，把文件夹设置为SourceRoot就不会报错了
os.environ['HADOOP_CONF_DIR']= "/export/server/hadoop/etc/hadoop"
改动2：在集群上运行，本地文件就不可以用了，需要用hdfs文件
# 在集群中运行，我们需要用HDFS路径，不能用本地路径
file_rdd = sc.textFile("hdfs://Tnode1:8020/input/order.text")
改动3：
"""
如果提交到集群运行，除了主代码以外，还依赖了其它的代码文件
需要设置一个参数，来告知spark，还有依赖文件要同步上传到集群中
参数叫做：spark.submit.pyFiles
参数的值可以是单个.py文件，也可以是.zip压缩包（有多个依赖文件的时候可以用zip压缩后上传）
"""
conf.set("spark.submit.pyFiles","defs_19.py")
完整代码：19_operators_runOnYarn.py
# coding:utf8
import json
from pyspark import SparkConf, SparkContext
import os
from defs_19 import city_with_category
# 导入自己写的函数时，把文件夹设置为SourceRoot就不会报错了
os.environ['HADOOP_CONF_DIR']= "/export/server/hadoop/etc/hadoop"
if __name__ == '__main__':
# 提交到yarn集群，master设置为yarn
conf = SparkConf().setAppName("SparkDemo01").setMaster("yarn")
"""
如果提交到集群运行，除了主代码以外，还依赖了其它的代码文件
需要设置一个参数，来告知spark，还有依赖文件要同步上传到集群中
参数叫做：spark.submit.pyFiles
参数的值可以是单个.py文件，也可以是.zip压缩包（有多个依赖文件的时候可以用zip压缩后上传）
"""
conf.set("spark.submit.pyFiles","defs_19.py")
sc = SparkContext(conf=conf)
# 在集群中运行，我们需要用HDFS路径，不能用本地路径
file_rdd = sc.textFile("hdfs://Tnode1:8020/input/order.text")
# 进行rdd数据的split 按照|符号进行，得到一个json数据
jsons_rdd = file_rdd.flatMap(lambda line: line.split("|"))
# 通过python内置的json库，完成json字符串到字典对象的转换
dict_rdd = jsons_rdd.map(lambda json_str: json.loads(json_str))
# 过滤数据，只保留北京的数据
beijing_rdd = dict_rdd.filter(lambda d: d['areaName'] == '北京')
# 组合北京和商品类型形成的字符串
category_rdd = beijing_rdd.map(city_with_category)
# 对结果集进行去重操作
result_rdd = category_rdd.distinct()
# 输出
print(result_rdd.collect())
依赖代码：defs_19.py
# coding:utf8
def city_with_category(data):
return data['areaName'] + '_' +data['category']
输出结果：
['北京_书籍', '北京_食品', '北京_服饰', '北京_平板电脑', '北京_家具', '北京_手机', '北京_家电', '北京_电脑']
在服务器上通过spark-submit 提交到集群运行
# --py-files 可以帮你指定你依赖的其它python代码，支持.zip(一堆)，也可以单个.py文件都行。
/export/server/spark/bin/spark-submit --master yarn --py-files ./defs.py ./main.py
服务器上程序运行结果：
注意，在服务器上跑时，需要把conf中的setMaster去掉
即conf = SparkConf().setAppName(“SparkDemo01”).setMaster(“yarn”)改为：
conf = SparkConf().setAppName(“SparkDemo01”)
2.5 常用Action算子
2.5.1 countByKey算子
功能：统计key出现的次数（一般适用于KV型的RDD）
代码演示：20_operators_countByKey.py
# coding:utf8
import json
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.textFile("../data/input/words.txt")
rdd2 = rdd.flatMap(lambda x:x.split(" ")).map(lambda x: (x, 1))
# 通过countByKey来对key进行计数，这是一个Action算子
result = rdd2.countByKey()
print(result)
print(list(result))
print(result["hello"])
print(type(result))
输出结果：
defaultdict(<class 'int'>, {'hello': 3, 'spark': 1, 'hadoop': 1, 'flink': 1})
['hello', 'spark', 'hadoop', 'flink']
3
<class 'collections.defaultdict'>
2.5.2 collect算子
功能：将RDD各个分区内的数据，统一收集到Driver中，形成一个List对象
用法：rdd.collect()
这个算子，是将RDD各个分区数据都拉取到Driver
注意的是，RDD是分布式对象，其数据量可以很大，
所以用这个算子之前要心知肚明地了解 结果数据集不会太大。
不然，会把Driver内存撑爆
2.5.3 reduce算子
功能：对RDD数据集按照你传入的逻辑进行聚合
语法：
rdd.reduce(func)
# func:(T,T)——>T
# 2参数传入1个返回值，返回值要和参数要求类型一致
代码演示：21_operators_reduce.py
# coding:utf8
import json
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([1,2,3,4,5])
print(rdd.reduce(lambda a, b: a + b))
输出结果：
15
2.5.4 fold算子
功能：和reduce一样，接收传入逻辑进行聚合，聚合是带有初始值的，
这个初始值聚合会作用在：
- 分区内聚合
- 分区间聚合
比如：[[1,2,3],[4,5,6],[7,8,9]]
数据量分布在3个分区
分区1： 1、2、3 聚合的时候带上10作为初始值得到16
分区3： 4、5、6 聚合的时候带上10作为初始值得到25
分区4： 7、8、9 聚合的时候带上10作为初始值得到34
3个分区的结果做聚合也带上初始值10，所以结果是10+16+25+34 = 85
代码演示：22_operators_fold.py
# coding:utf8
import json
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([1,2,3,4,5,6,7,8,9],3)
print(rdd.glom().collect())
print(rdd.fold(10, lambda a, b: a + b))
输出结果：
[[1, 2, 3], [4, 5, 6], [7, 8, 9]]
85
2.5.5 first算子
功能：取出RDD的第一个元素
用法
sc.parallelize([3,2,1]).first()
输出：3
2.5.6 take算子
功能：取RDD的前N个元素。组合成list返回给你
用法：
>>> sc.parallelize([3,2,1,4,5,6]).take(5)
[3, 2, 1, 4, 5]
2.5.7 top算子
功能：对RDD数据集进行降序排序，取前N个
用法：
>>> sc.parallelize([3,2,1,4,5,6]).top(3) # 表示取降序前3个
[6, 5, 4]
2.5.8 count算子
功能：计算RDD有多少条数据，返回值是一个数字
用法：
>>> sc.parallelize([3,2,1,4,5,6]).count()
6
2.5.9 takeSample算子
功能：随机抽样RDD的数据
用法：
takeSample(参数1：True or False，参数2：采样数，参数3：随机数种子)
- 参数1：True表示允许取同一个数据，False表示不允许取同一个数据，和数据内容无关，是否重复表示的是同一个位置的数据（有、无放回抽样）
- 参数2：抽样要几个
- 参数3：随机数种子，这个参数传入一个数字即可，随意给
随机数种子 数字可以随便传，如果传同一个数字 那么取出的结果是一致的。
一般参数3 我们不传，Spark会自动给与随机的种子。
代码演示：23_operators_takeSample.py
# coding:utf8
import json
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([1, 3, 5, 3, 1, 3, 2, 6, 7, 8, 6],1)
result = rdd.takeSample(False,5,1)
# 随机抽样可以抽出相同的数据，只是位置不同而已
# 随机数种子能让随机数不再继续发生变化
print(result)
输出结果：
[2, 7, 6, 6, 3]
注意：
随机抽样可以抽出相同的数据，只是位置不同而已
随机数种子能让随机数不再继续发生变化
2.5.10 takeOrdered
功能：对RDD进行排序取前N个
用法：
rdd.takeOrdered(参数1，参数2)
- 参数1 要几个数据
- 参数2 对排序的数据进行更改（不会更改数据本身，只是在排序的时候换个样子）
这个方法按照元素自然顺序升序排序，如果你想玩倒叙，需要参数2 来对排序的数据进行处理
代码演示：24_operators_takeOrdered.py
# coding:utf8
import json
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([1, 3, 2, 4, 7, 9, 6], 1)
print(rdd.takeOrdered(3))
print(rdd.takeOrdered(3,lambda x:-x))
输出结果：
[1, 2, 3]
[9, 7, 6]
2.5.11 foreach算子
功能：对RDD的每一个元素，执行你提供的逻辑的操作（和map一个思想），但是这个方法没有返回值
用法：
rdd.foreach(func)
# func:(T) ——> None
代码演示：25_operators_foreach.py
# coding:utf8
import json
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([1, 3, 2, 4, 7, 9, 6], 1)
rdd.foreach(lambda x: print(x * 10))
输出结果：
10
30
20
40
70
90
60
2.5.12 saveAsTextFile
功能：将RDD的数据写入文本文件中
支持本地写出，hdfs等文件系统
代码演示：
# coding:utf8
import json
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([1,3,2,4,7,9,6],3)
rdd.saveAsTextFile("hdfs://Tnode1:8020/test/output/out1")
运行结果：
注意：保存文件API，是分布式执行的
这个API的执行数据是不经过driver的
如图，写出的时候，每个分区所在的Executor直接控制数据写出到目标文件系统中
所有才会一个分区产生一个结果文件
2.5.13 注意点
我们学习的action中：
- foreach
- saveAsTextFile
这两个算子是分区（Executor）直接执行的，跳过Driver，由分区所在的Executor直接执行
反之：其余的Action算子都会将结果发送至Driver
2.6 分区操作算子
2.6.1 mapPartitions算子
transformation算子
图解：
如图，mapPartition一次被传递的是一整个分区的数据
作为一个迭代器（一次性list）对象传入过来。
代码演示：27_operators_mapPartitions.py
# coding:utf8
import json
from pyspark import SparkConf, SparkContext
import time
if __name__ == '__main__':
start_time = time.time()
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
# 效果和map一样，但是性能比map好，cpu计算没有省，但是网络IO少很多
rdd = sc.parallelize([1,3,2,4,7,9,6],3)
def process(iter):
result = []
for it in iter:
result.append(it*10)
return result
print(rdd.mapPartitions(process).collect())
# print(rdd.map(lambda x:x*10).collect())
end_time = time.time()
gap_time = (end_time - start_time)
gap_time = round(gap_time, 4) # 保留四位小数
print("执行本程序共耗时：" + str(gap_time) + "s")
输出结果：
[10, 30, 20, 40, 70, 90, 60]
执行本程序共耗时：8.0515s
注意：效果和map一样，但是性能比map好，cpu计算没有省，但是网络IO少很多
2.6.2 foreachPartition算子
Action算子
功能：和普通foreach一致，一次处理的是一整个分区数据
代码演示：28_operators_foreachPartitions.py
# coding:utf8
import json
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([1,3,2,4,7,9,6],3)
def process(iter):
result = []
for it in iter:
result.append(it*10)
print(result)
rdd.foreachPartition(process)
输出结果：
[70, 90, 60]
[10, 30]
[20, 40]
foreachPartition 就是一个没有返回值的mapPartitions
2.6.3 partitionBy算子
transformation算子
功能：对RDD进行自定义分区操作
用法：
rdd.partitionBy(参数1，参数2)
- 参数1 重新分区后有几个分区
- 参数2 自定义分区规则，函数传入
参数2：(K)——>int
一个传入参数进来，类型无所谓，但是返回值一定是int类型，
将key传给这个函数，你自己写逻辑，决定返回一个分区编号
分区编号从0开始，不要超出分区数-1
代码演示：29_operators_partitionBy.py
# coding:utf8
import json
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([('hadoop', 1), ('spark', 1), ("hello", 1), ("flink", 1), ("hadoop", 1), ("spark", 1)])
# 使用partitionBy 自定义 分区
def process(k):
if 'hadoop' == k or 'hello' == k: return 0
if 'spark' == k: return 1
return 2
print(rdd.partitionBy(3, process).glom().collect())
输出结果：分区依次为0、1、2
[[('hadoop', 1), ('hello', 1), ('hadoop', 1)], [('spark', 1), ('spark', 1)], [('flink', 1)]]
分区号不要超标，你设置3个分区，分区号只能是0 1 2
设置5个分区 分区号只能是0 1 2 3 4
2.6.4 repartition算子
transformation算子
功能：对RDD的分区执行重新分区（仅数量）
用法：
rdd.repartition(N)
传入N 决定新的分区数
代码演示：30_operators_repartition_and_coalesce.py
# coding:utf8
import json
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([1, 2, 3, 4, 5], 3)
# repartition 修改分区
print(rdd.repartition(1).getNumPartitions())
print(rdd.repartition(5).getNumPartitions())
# coalesce 修改分区
print(rdd.coalesce(1).getNumPartitions())
print(rdd.coalesce(5,shuffle=True).getNumPartitions())
输出结果：
1
5
1
5
注意：对分区的数量进行操作，一定要慎重
一般情况下，我们写spark代码除了要求全局排序设置为1个分区外，
多数时候，所有API中关于分区相关的代码我们都不太理会
因为，如果你改分区了
- 会影响并行计算（内存迭代的并行管道数量）
后面学- 分区如果增加，极大可能导致shuffle
2.6.5 coalesce算子
transformation算子
功能：对分区进行数量增减
用法：
rdd.coalesce(参数1，参数2)
- 参数1，分区数
- 参数2，True or False
True表示允许shuffle，也就是可以加分区
False表示不允许shuffle，也就是不能加分区，False是默认
代码见2.6.4
对比repartition，一般使用coalesce较多，因为加分区要写参数2
这样避免写repartition的时候手抖了加分区了
2.6.6 mapValues算子
Transformation算子
功能：针对二元元组RDD，对其内部的二元元组的Value执行map操作
语法：
rdd.mapValues(func)
# func: (V)——> U
# 注意，传入的参数，是二元元组的 value值
# 我们这个传入的方法，只对value进行处理
代码演示：
# coding:utf8
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setAppName("create rdd").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([('a', 1), ('a', 11), ('a', 6), ('b', 3), ('b', 5)])
# rdd.map(lambda x:(x[0],x[1]*10))
# 将二元元组的所有value都乘以10进行处理
print(rdd.mapValues(lambda x: x * 10).collect())
输出结果：
[('a', 10), ('a', 110), ('a', 60), ('b', 30), ('b', 50)]
2.6.7 join算子
Transformation算子
功能：对两个RDD执行join操作（可以实现SQL的内、外连接）
注意：join算子只能用于二元元组
代码见 2.4.9
2.7 面试题
groupByKey和reduceByKey的区别
在功能上的区别：
-
groupByKey仅仅只有分组功能而已 -
reduceByKey除了有ByKey的分组功能外，还有reduce聚合功能，所以是一个分组+聚合一体化的算子
如果对数据执行分组+聚合，那么使用这2个算子的性能差别是很大的
reduceByKey的性能是远大于：groupByKey+聚合逻辑的
因为：
如图，这是groupByKey+聚合逻辑的执行流程。
因为，groupByKey只能分组，所以，执行上是先分组（shuffle）后聚合
再来看reduceByKey：
如图，reduceByKey由于自带聚合逻辑，所以可以完成：
- 先在分区内做预聚合
- 然后再走分组流程（shuffle）
- 分组后再做最终聚合
对于groupByKey，reduceByKey最大的提升在于，分组前进行了预聚合，那么在shuffle分组节点，被shuffle的数据可以极大地减少
这就极大地提升了性能
分组+聚合，首选
reduceByKey，数据越大，对groupByKey的优势就越高
2.8 总结
- RDD创建方式有哪几种方法？
通过并行化集合的方式（本地集合转分布式集合）
或者读取数据的方式创建（TextFile、WholeTextFile）
- RDD分区数如何查看？
通过getNumPartitions API查看，返回值Int
-
Transformation和Action的区别？
转换算子的返回值100%是RDD，而Action算子的返回值100%不是RDD转换算子是懒加载的，只有遇到Action才会执行，Action就是转换算子处理链条的开关。
-
哪两个Action算子的结果不经过Driver，直接输出？
foreach和saveAsTextFile 直接由Executor执行后输出，不会将结果发送到Driver上去
-
reduceByKey和groupByKey的区别？
reduceByKey自带聚合逻辑，groupByKey不带
如果做数据聚合reduceByKey的效率更好，因为可以先聚合后shuffle在最终聚合，传输的IO小
-
mapPartitions和foreachPartition的区别？
mapPartitions带有返回值 foreachPartition不带
-
对于分区操作有什么要注意的地方？
尽量不要增加分区，可能破坏内存迭代的计算管道
3. RDD的持久化
3.1 RDD的数据是过程数据
RDD之间进行相互迭代计算（Transformation的转换），当执行开启后，新的RDD生成，代表老RDD的消失。
RDD的数据是过程数据，只在处理的过程中存在，一旦处理完成，就不见了。
这个特性可以最大化地利用资源，老旧RDD没用了 就从内存中清理，给后续的计算腾出内存空间。
如上图，rdd3被2次使用，第一次使用之后，其实RDD3就不存在了。
第二次使用的时候，只能基于RDD的血缘关系，从RDD1重新执行，构建出来RDD3，供RDD5使用。
3.2 RDD的缓存
3.2.1 缓存
对于上述的场景，肯定要执行优化，优化就是：
RDD3如果不消失，那么RDD1——>RDD2——>RDD3这个链条就不会执行2次，或者更多次
RDD的缓存技术：Spark提供了缓存API，可以让我们通过调用APi，将指定的RDD数据保留在内存或者硬盘上
缓存的API
# RDD3 被2次使用，可以加入缓存进行优化
rdd3.cache() # 缓存到内存中
rdd3.persist(StorageLevel.MEMORY_ONLY) # 仅内存缓存
rdd3.persist(StorageLevel.MEMORY_ONLY_2) # 仅内存缓存,2个副本
rdd3.persist(StorageLevel.DISK_ONLY) # 仅缓存硬盘上
rdd3.persist(StorageLevel.DISK_ONLY_2) # 仅缓存硬盘上,2个副本
rdd3.persist(StorageLevel.DISK_ONLY_3) # 仅缓存硬盘上,3个副本
rdd3.persist(StorageLevel.MEMORY_AND_DISK) # 先放内存，不够放硬盘
rdd3.persist(StorageLevel.MEMORY_AND_DISK_2) # 先放内存，不够放硬盘，2个副本
rdd3.persist(StorageLevel.OFF_HEAP) # 堆外内存（系统内存）
# 如上API，自行选择使用即可
# 一般建议使用rdd3.persist(StorageLevel.MEMORY_AND_DISK)
# 如果内存比较小的集群，建议使用rdd3.persist(StorageLevel.DISK_ONLY)或者就别用缓存了 用CheckPoint
# 主动清理缓存的API
rdd.unpersist()
3.2.2 缓存特点
- 缓存技术可以将过程RDD数据，持久化保存到内存或者硬盘上
- 但是，这个保存在设定上是认为不安全的。
缓存的数据在设计上是认为有丢失风险的。
所以，缓存有一个特点就是：其保留RDD之间的血缘（依赖）关系
一旦缓存丢失，可以基于血缘关系的记录，重新计算这个RDD的数据
缓存如何丢失：
- 在内存中的缓存是不安全的，比如断电、计算任务内存不足，把缓存清理给计算让路
- 硬盘中因为硬盘损坏也是可能丢失的。
代码演示：31_cache.py
# coding:utf8
from pyspark.storagelevel import StorageLevel
from pyspark import SparkConf, SparkContext
import time
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd1 = sc.textFile("../data/input/words.txt")
rdd2 = rdd1.flatMap(lambda x: x.split(" "))
rdd3 = rdd2.map(lambda x: (x, 1))
# 给rdd3加缓存
# rdd3.cache()
rdd3.persist(StorageLevel.MEMORY_AND_DISK_2) # 设置缓存级别
rdd4 = rdd3.reduceByKey(lambda a, b: a + b)
result = rdd4.collect()
print(result)
rdd5 = rdd3.groupByKey()
rdd6 = rdd5.mapValues(lambda x:sum(x))
print(rdd6.collect())
# 取消缓存
rdd3.unpersist()
time.sleep(10000000)
输出结果：
[('hadoop', 1), ('hello', 3), ('spark', 1), ('flink', 1)]
[('hadoop', 1), ('hello', 3), ('spark', 1), ('flink', 1)]
3.2.3 缓存是如何保存的
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-EP3ykOn3-1692435339403)(https://cdn.jsdelivr.net/gh/Sql88/BlogImg@main/img/%E7%BC%93%E5%AD%98%E6%98%AF%E5%A6%82%E4%BD%95%E4%BF%9D%E5%AD%98%E7%9A%842.png)]
如图，RDD是将自己分区的数据，每个分区自行将其数据保存在其所在的Executor内存和硬盘上。
这是分散存储
3.3 RDD的CheckPoint
3.3.1 RDD CheckPoint
CheckPoint技术，也是将RDD的数据，保存起来。
但是它仅支持硬盘存储
并且：
- 它被设计认为是安全的
- 不保留
血缘关系
3.3.2 CheckPoint是如何保存数据的
如图：CheckPoint存储RDD数据，是集中收集各个分区数据进行存储。而缓存是分散存储
3.3.3 缓存和CheckPoint的对比
- CheckPoint不管分区数量多少，风险是一样的，缓存分区越多，风险越高
- CheckPoint支持写入HDFS，缓存不行，HDFS是高可靠存储，CheckPoint被认为是安全的
- CheckPoint不支持内存，缓存可以，缓存如果写内存，性能比CheckPoint要好一些
- CheckPoint因为设计是安全的，所以不保留血缘关系，而缓存因为设计上认为不安全，所以保留
3.3.4 代码
# 设置CheckPoint第一件事情，选择CP的保存路径
# 如果是Local模式，可以支持本地文件系统，如果在集群运行，千万要用HDFS
sc.setCheckpointDir("hdfs://node1:8020/output/bj52ckp")
# 用的时候，直接调用checkPoint算子即可。
rdd.checkpoint()
完整代码演示：32_checkPoint.py
# coding:utf8
import json
from pyspark.storagelevel import StorageLevel
from pyspark import SparkConf, SparkContext
import time
if __name__ == '__main__':
conf = SparkConf().setAppName("test").setMaster("local[*]")
sc = SparkContext(conf=conf)
# 1.告知spark，开启checkPoint功能
sc.setCheckpointDir("hdfs://Tnode1:8020/output/ckp")
rdd1 = sc.textFile("../data/input/words.txt")
rdd2 = rdd1.flatMap(lambda x: x.split(" "))
rdd3 = rdd2.map(lambda x: (x, 1))
# 调用checkPoint API 保存数据即可
rdd3.checkpoint()
rdd4 = rdd3.reduceByKey(lambda a, b: a + b)
result = rdd4.collect()
print(result)
rdd5 = rdd3.groupByKey()
rdd6 = rdd5.mapValues(lambda x:sum(x))
print(rdd6.collect())
# 取消缓存
rdd3.unpersist()
time.sleep(10000000)
输出结果：
[('hadoop', 1), ('hello', 3), ('spark', 1), ('flink', 1)]
[('hadoop', 1), ('hello', 3), ('spark', 1), ('flink', 1)]
3.3.5 注意
CheckPoint是一种重量级的使用，也就是RDD的重新计算成本很高的时候，我们采用CheckPoint比较合适。
或者数据量很大，用CheckPoint比较合适。
如果数据量小，或者RDD重新计算是非常快的，用CheckPoint没啥必要
Cache和CheckPoint两个API都不是Action类型
所以，想要它俩工作，必须在后面接上Action
接上Action的目的，是让RDD有数据，而不是为了CheckPoint和Cache工作。
3.3.6 总结
1.Cache和CheckPoint的区别
- Cache是轻量化保存RDD数据，可存储在内存和硬盘，是分散存储，设计上数据是不安全的（保留RDD血缘关系）
- CheckPoint是重量级保存RDD数据，是集中存储，只能存储在硬盘（HDFS）上，设计上是安全的（不保留RDD血缘关系）
2.Cache和CheckPoint的性能对比？
- Cache性能更好，因为是分散存储，各个Executor并行执行，效率高，可以保存到内存中（占内存），更快
- CheckPoint比较慢，因为是集中存储，涉及到网络IO，但是存储到HDFS上更加安全（多副本）
4. Spark案例练习
4.1 搜索引擎日志分析案例
数据格式：
需求：
- 用户搜索的关键词分析
- 用户和关键词组合分析
- 热门搜索时间段分析
案例实现代码：
# coding:utf8
from pyspark import SparkConf, SparkContext
from pyspark.storagelevel import StorageLevel
import jieba
from operator import add
def context_jieba(data):
"""通过jieba分词工具 进行分词操作"""
seg = jieba.cut_for_search(data)
l = []
for word in seg:
l.append(word)
return l
def filter_words(data):
"""过滤不要的 谷、帮、客 湖"""
return data not in ['谷', '帮', '客', '湖']
def append_words(data):
"""修订某些关键词的内容"""
if data == '传智播': data = '传智播客'
if data == '院校': data = '院校帮'
if data == '博学': data = '博学谷'
if data == '数据': data = '数据湖'
return (data, 1)
def extract_user_and_word(data):
"""传入数据是 元组(1,我喜欢传智播客)"""
user_id = data[0]
content = data[1]
# 对content进行分词
words = context_jieba(content)
return_list = []
for word in words:
# 不要忘记过滤 \谷\帮\客\湖
if filter_words(word):
return_list.append((user_id + '_' + append_words(word)[0], 1))
return return_list
if __name__ == '__main__':
conf = SparkConf().setAppName("SparkDemo2")
sc = SparkContext(conf=conf)
# 1.读取文件
file_rdd = sc.textFile("hdfs://Tnode1/input/SogouQ.txt")
# 2. 对数据进行切分 \t
split_rdd = file_rdd.map(lambda x: x.split("\t"))
# 3. 因为要做多个需求，split_rdd 作为基础的rdd 会被多次使用
split_rdd.persist(StorageLevel.DISK_ONLY)
# TODO:需求1：用户搜索的关键‘词’分析
# 主要分析热点词
# 将所有的搜索内容取出
# print(split_rdd.takeSample(True, 3))
context_rdd = split_rdd.map(lambda x: x[2])
# 对搜索的内容进行分词分析
words_rdd = context_rdd.flatMap(context_jieba)
# print(words_rdd.collect())
# 异常的数据：
# 数据 湖 ——> 数据湖
# 院校 帮 ——> 院校帮
# 博学 谷 ——> 博学谷
# 传智播 客——> 传智播客
filtered_rdd = words_rdd.filter(filter_words)
# 将关键词转换：传智播 --> 传智播客
final_words_rdd = filtered_rdd.map(append_words)
# 对单词进行分组、聚合、排序 求出前五名
result1 = final_words_rdd.reduceByKey(lambda a, b: a + b). \
sortBy(lambda x: x[1], ascending=False, numPartitions=1). \
take(5)
print("需求1结果：", result1)
# TODO：需求2：用户和关键词组合分析
# 1，我喜欢传智播客
# 1 + 我 1+喜欢 1+传智播客
user_content_rdd = split_rdd.map(lambda x: (x[1], x[2]))
# 对用户的搜索内容进行分词，分词后和用户ID再次组合
user_word_with_one_rdd = user_content_rdd.flatMap(extract_user_and_word)
# 对内容进行分组、聚合、排序、求前5
result2 = user_word_with_one_rdd.reduceByKey(lambda a, b: a + b). \
sortBy(lambda x: x[1], ascending=False, numPartitions=1). \
take(5)
print("需求2结果：", result2)
# TODO:需求3：热门搜索时间段分析
# 取出来所有的时间
time_rdd = split_rdd.map(lambda x: x[0])
# 对时间进行处理，只保留小时精度即可
hour_with_one_rdd = time_rdd.map(lambda x: (x.split(":")[0], 1))
# 分组、聚合、排序
result3 = hour_with_one_rdd.reduceByKey(add). \
sortBy(lambda x: x[1], ascending=False, numPartitions=1). \
collect()
print("需求3结果：", result3)
输出结果：
需求1结果： [('scala', 2310), ('hadoop', 2268), ('博学谷', 2002), ('传智汇', 1918), ('itheima', 1680)]
需求2结果： [('6185822016522959_scala', 2016), ('41641664258866384_博学谷', 1372), ('44801909258572364_hadoop', 1260), ('7044693659960919_仓库', 1120), ('15984948747597305_传智汇', 1120)]
需求3结果： [('20', 3479), ('23', 3087), ('21', 2989), ('22', 2499), ('01', 1365)
4.2 提交到集群运行
# 普通提交
/export/server/spark/bin/spark-submit --master yarn SparkDemo2.py
# 压榨集群式提交
# 每个executor吃14g内存，8核cpu，总共3个executor
/export/server/spark/bin/spark-submit --master yarn --executor-memory 14g --executor-cores 8 --num-executors 3 ./SparkDemo2.py
输出结果：
要注意代码中：
- master部分删除
- 读取的文件路径改为hdfs才可以
4.3 作业
代码演示：
# coding:utf8
from pyspark import SparkContext, StorageLevel
from pyspark import SparkConf
if __name__ == '__main__':
conf = SparkConf().setAppName("sparkHomeWork01").setMaster("local[*]")
sc = SparkContext(conf=conf)
file_rdd = sc.textFile("../../data/input/apache.log")
file_rdd.persist(StorageLevel.MEMORY_AND_DISK_2)
# 需求1：TODO：计算当前网站访问的PV（被访问次数）
visit_Num = file_rdd.count()
print("当前网站的被访问次数：", visit_Num) # 14
# 需求2：TODO：当前网站访问的用户数
userNum = file_rdd.distinct().count()
print("当前网站的访问用户数：", userNum) #
# 需求3：TODO：有哪些IP访问了本网站？
Ip_rdd1 = file_rdd.map(lambda x: x.split(" "))
Ip_rdd1.cache()
Ip_rdd2 = Ip_rdd1.map(lambda x: x[0]).distinct()
# print(IP_rdd2.collect())
print("有哪些IP访问了本网站：", Ip_rdd2.collect())
# 需求4 TODO：哪个页面访问量最高
page_rdd1 = Ip_rdd1.map(lambda x:x[-1])
page_rdd2 = page_rdd1.map(lambda x:(x,1))
page_rdd3 = page_rdd2.reduceByKey(lambda a,b:a+b)
# page = page_rdd3.sortBy(lambda x:x[1],ascending=False,numPartitions=1).take(1)
page = page_rdd3.takeOrdered(1,lambda x:-x[1])
page = page[0]
print(page)
print("访问量最高的页面是：",page[0],"共被访问：",page[1],"次")
输出结果：sparkHomeWork01.py
当前网站的被访问次数： 14
当前网站的访问用户数： 9
有哪些IP访问了本网站： ['83.149.9.216', '10.0.0.1', '86.149.9.216']
('/presentations/logstash-monitorama-2013/css/print/paper.css', 13)
访问量最高的页面是： /presentations/logstash-monitorama-2013/css/print/paper.css 共被访问： 13 次
5. 共享变量
5.1 广播变量
5.1.1 问题引出
有如下代码：
上述代码，本地list对象和分布式对象RDD有了关联。如下图：
本地list对象，被发送到每个分区的处理线程上使用，也就是一个executor内，其实存放了2份一样的数据。
executor是进程，进程内资源共享，这2份数据没有必要，造成了内存浪费。
5.1.2 解决方案-广播变量
如果本地list对象标记为广播变量对象，那么
当上述场景出现的时候，Spark只会：
- 给每个Executor来一份数据，而不像原本那样，每一个分区的处理线程都来一份，节省内存。
如图，使用广播变量后，每个Executor只会收到一份数据集。
内部的各个线程（分区）共享这一份数据集。
使用方式：
# 1. 将本地list 标记成广播变量即可
broadcast = sc.broadcast(stu_info_list)
# 2. 使用广播变量，从broadcast对象中取出本地list对象即可
value = broadcast.value
# 也就是 先放进去broadcast内部，然后从broadcast内部在取出来用，中间传输的是broadcast这个对象了
# 只要中间传输的是broadcast对象，spark就会留意，只会给每个Executor发一份了，而不是傻傻的哪个分区都要给
代码演示：33_broadcast.py
# coding:utf8
import json
from pyspark.storagelevel import StorageLevel
from pyspark import SparkConf, SparkContext
import time
if __name__ == '__main__':
conf = SparkConf().setAppName("33_broadcast.py").setMaster("local[*]")
sc = SparkContext(conf=conf)
stu_info_list = [(1, '张大仙', 11),
(2, '王晓晓', 13),
(3, '张甜甜', 11),
(4, '王大力', 11)]
# 1.将本地Python List对象标记为广播变量
broadcast = sc.broadcast(stu_info_list)
score_info_rdd = sc.parallelize([
(1, '语文', 99),
(2, '数学', 99),
(3, '英语', 99),
(4, '编程', 99),
(1, '语文', 99),
(2, '编程', 99),
(3, '语文', 99),
(4, '英语', 99),
(1, '语文', 99),
(3, '英语', 99),
(2, '编程', 99)
])
def map_func(data):
name = ''
id = data[0]
# 匹配本地list和分布式rdd中的学生ID 匹配成功后 即可获得当前学生的姓名
# 2.在使用到本地集合对象的地方，从广播变量中取出来用即可
for stu_info in broadcast.value:
if stu_info[0] == id:
name = stu_info[1]
break
return (name, data[1], data[2])
print(score_info_rdd.map(map_func).collect())
"""
广播变量使用场景：本地集合对象和 分布式集合对象(RDD) 进行关联的时候
需要将本地集合对象封装为广播变量
可以节省：
1. 网络IO的次数
2. Executor的内存占用
"""
输出结果：
[('张大仙', '语文', 99), ('王晓晓', '数学', 99), ('张甜甜', '英语', 99), ('王大力', '编程', 99), ('张大仙', '语文', 99), ('王晓晓', '编程', 99), ('张甜甜', '语文', 99), ('王大力', '英语', 99), ('张大仙', '语文', 99), ('张甜甜', '英语', 99), ('王晓晓', '编程', 99)]
5.2 累加器
5.2.1 需求
想要对map算子计算中的数据，进行数据累加，得到全部数据计算完后的累加结果
5.2.2 没有累加器的代码演示
# coding:utf8
from pyspark import SparkConf, SparkContext
# 演示spark的accumulator累加器
if __name__ == '__main__':
conf = SparkConf().setAppName("create rdd").setMaster("local[*]")
sc = SparkContext(conf=conf)
rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10],2)
count = 0
def map_func(data):
global count
count += 1
print(count)
rdd.map(map_func).collect()
print(count)
# 代码中count 最后打印结果是0
输出结果：
1
2
3
4
5
1
2
3
4
5
0
代码的问题在于：
- count来自driver对象，当在分布式的map算子中需要count对象的时候，driver会将count对象发送给每一个executor一份（复制发送）
- 每个executor各自收到一个，在最后执行print(count) 的时候，这个被打印的count依旧是driver那个
- 所以，不管executor中累加到多少，都和driver这个count无关
5.2.3 解决方法-累加器
代码演示：
# coding:utf8
import json
from pyspark.storagelevel import StorageLevel
from pyspark import SparkConf, SparkContext
import time
if __name__ == '__main__':
conf = SparkConf().setAppName("33_broadcast.py").setMaster("local[*]")
sc = SparkContext(conf=conf)
# 10条数据 2个分区
rdd = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], 2)
# count = 0
# Spark 提供的累加器变量，参数是初始值
acmlt = sc.accumulator(0)
def map_func(data):
global acmlt
acmlt += 1
print(acmlt)
rdd.map(map_func).collect()
# rdd2 = rdd.map(map_func)
# rdd2.cache()
# rdd2.collect()
# rdd3 = rdd2.map(lambda x:x)
# rdd3.collect()
print(acmlt) # 结果是10
"""
累加器使用的注意点：
某个rdd使用完后，再被新的rdd重新调用，有可能会产生和想象中不一样的结果
避免方法：给需要重用的rdd加缓存
"""
如上代码，将全部的count对象，都替换成acmlt对象即可
这个对象就是累加器对象，构建方式：sc.accumulator(初始值)即可构建。
这个对象唯一和前面提到的count不同的是，这个对象可以从各个Executor中收集它们的执行结果，作用回自己身上。
输出结果：
1
2
3
4
5
1
2
3
4
5
10
5.2.4 累加器的注意事项
如上代码，第一次rdd2被action后，累加器值是10，然后rdd2就没有了（没数据了）
当rdd3构建出来的时候，是依赖rdd2，rdd2没数据，那么rdd2就要重新生成
重新生成就导致累加器累加数据的代码再次被执行，
所以代码的结果是20
也就是说，使用累加器的时候，要注意，因为rdd是过程数据，如果rdd被多次使用
可能重新构建此rdd
如果累加器累加代码，存在重新构建的步骤中
累加器累加代码就可能被多次执行。
如何解决：加缓存或者CheckPoint即可
5.3 综合案例
5.3.1 需求
对上面的数据执行：
- 正常的单词进行单词计数
- 特殊字符统计出现有多少个
特殊字符定义如下：
abnormal_char = [",",".","!","#","$","%"]
代码演示：
# coding:utf8
import json
from pyspark.storagelevel import StorageLevel
from pyspark import SparkConf, SparkContext
import time
import re
if __name__ == '__main__':
conf = SparkConf().setAppName("35_demo.py").setMaster("local[*]")
sc = SparkContext(conf=conf)
# 1.读取数据文件
file_rdd = sc.textFile("../data/input/accumulator_broadcast_data.txt")
# 特殊字符的List定义
abnormal_char = [',', '.', '!', '#', '$', '%']
# 2. 特殊字符list 包装成广播变量
broadcast = sc.broadcast(abnormal_char)
# 3.对特殊字符出现次数做累加，累加使用累加器最好
acmlt = sc.accumulator(0)
# 4.数据处理，先处理数据的空行,在Python中有内容就是True None就是False
lines_rdd = file_rdd.filter(lambda line: line.strip())
# 5.去除前后的空格
data_rdd = lines_rdd.map(lambda line: line.strip())
# 6.对数据进行切分，按照正则表达式切分，因为空格分隔符某些单词之间是两个或多个空格
# 正则表达式 \s+ 表示 不确定多少个空格，最少一个空格
words_rdd = data_rdd.flatMap(lambda line: re.split("\s+", line))
# 7. 当前words_rdd中有正常单词，也有特殊符号
# 现在需要过滤数据，保留正常单词用于单词计数，在过滤的过程中 对特殊符号做计数
def filter_func(data):
global acmlt
# 取出广播变量中存储的特殊符号lsit
abnormal_chars = broadcast.value
if data in abnormal_chars:
# 表示这个是特殊字符
acmlt += 1
return False
else:
return True
normal_words_rdd = words_rdd.filter(filter_func)
# 8. 正常单词的单词计数逻辑
result_rdd = normal_words_rdd.map(lambda x: (x, 1)). \
reduceByKey(lambda a, b: a + b)
print("正常单词计数结果：",result_rdd.collect())
print("特殊字符数量：",acmlt)
输出结果：
正常单词计数结果： [('hadoop', 3), ('hive', 6), ('hdfs', 2), ('spark', 11), ('mapreduce', 4), ('sql', 2)]
特殊字符数量： 8
5.4 总结
-
广播变量解决了什么问题？
分布式集合RDD和本地集合进行关联使用的时候，降低内存占用以及减少网络IO传输，提高性能。
-
累加器解决了什么问题
分布式代码执行中，进行全局累加
6.Spark内核调度（重点理解）
6.1 DAG
6.1.1 DAG
Spark的核心是根据RDD来实现的，Spark Scheduler则为Spark核心实现的重要一环，其作用就是任务调度。Spark
的任务调度就是如何组织任务去处理RDD中每个分区的数据，根据RDD的依赖关系构建DAG，基于DAG划分Stage，
将每个Stage中的任务发到指定节点运行。基于Spark的任务调度原理，可以合理规划资源利用，做到尽可能用最少的
资源高效地完成任务计算。
以词频统计WordCount程序为例，DAG图：
DAG：有向无环图（拓扑结构）
有向：有方向
无环：没有闭环
DAG：有方向没有形成闭环的一个执行流程图
比如：
此图，就是一个典型的DAG图。
有方向：RDD1——>RDD2——>…——>collect结束
无闭环：以action(collect) 结束了，没有形成闭环循环
作用：标识代码的逻辑执行流程
6.1.2 Job和Action
Action：返回值不是RDD的算子
它的作用是一个触发开关，会将action算子之前的一串rdd依赖执行起来
如图，我们前面写的搜索引擎日志分析案例中，前两个需求就是2个action，就产生了2个DAG
结论：
1个Action会产生1个DAG，如果在代码中有3个Action就产生3个DAG
一个Action产生的一个DAG，会在程序运行中产生一个
JOB所以：
1个ACTION = 1个DAG = 1个JOB如果一个代码中，写了3个Action，那么这个代码运行起来产生3个JOB，每个JOB有自己的DAG
一个代码运行起来，在Spark中称之为：
Application层级关系：
1个Application中，可以有多个JOB，每一个JOB内含一个DAG，同时每一个JOB都是由一个Action产生的。
6.1.3 DAG和分区
DAG是Spark代码的逻辑执行图，这个DAG的最终作用是；为了构建物理上的Spark详细执行计划而生。
所以，由于Spark是分布式（多分区）的，那么DAG和分区之间也是有关联的。
rdd1 = sc.textFile()
rdd2 = rdd1.flatMap()
rdd3 = rdd2.map()
rdd4 = rdd3.reduceByKey()
rdd4.action()
假设，全部RDD都是3个分区在执行
如图，就得到了带有分区关系的DAG图
6.2 DAG的宽窄依赖和阶段划分
在Spark RDD前后之间的关系为：
- 窄依赖
- 宽依赖
窄依赖：父RDD的一个分区，全部将数据发送给子RDD的一个分区
宽依赖：父RDD的一个分区，将数据发送给子RDD的多个分区
宽依赖还有一个别名：shuffle
6.2.1 窄依赖
6.2.2 宽依赖
6.2.3 阶段划分
对于Spark来说，会根据DAG，按照宽依赖，划分不同的DAG阶段
划分依据：从后向前，遇到宽依赖就划分出一个阶段，称之为stage
如图，可以看到，在DAG中，基于宽依赖，将DAG划分成了2个stage
在stage的内部，一定都是：窄依赖
6.3 内存迭代计算
如图，基于带有分区的DAG以及阶段划分。可以从图中得到 逻辑上最优的task分配，一个task是一个线程来具体执行
那么如上图，task1中rdd1 rdd2 rdd3的迭代计算，都是由一个task(线程完成)，这一阶段的这一条线，是纯内存计算。
如上图，task1 task2 task3，就形成了三个并行的 内存计算管道。
Spark默认受到全局并行度的限制，除了个别算子有特殊分区情况，大部分的算子，都会遵循全局并行度的要求，来规划自己的分区数。
如果全局并行度是3，其实大部分算子分区都是3
注意：Spark ，我们一般推荐只设置全局并行度，不要在算子上设置并行度。
除了一些排序算子外，计算算子就让他默认开分区就可以了
6.3.1 面试题
面试题1：Spark是怎么做内存计算的？DAG的作用？Stage阶段划分的作用？
- Spark会产生DAG图
- DAG图会基于分区和宽窄依赖关系划分阶段
- 一个阶段的内部都是窄依赖，窄依赖内，如果形成前后1:1的分区对应关系，就可以产生许多内存迭代计算的管道
- 这些内存迭代计算的管道，就是一个个具体的执行Task
- 一个Task是一个具体的线程，任务跑在一个线程内，就是走内存计算了。
面试题2：Spark为什么比MapReduce快
- Spark的算子丰富，MapReduce算子匮乏（Map和Reduce），MapReduce这个编程模型，很难在一套MR中处理复杂的任务。很多复杂任务，是需要写多个MapReduce进行串联。多个MR串联通过磁盘交互数据。
- Spark可以执行内存迭代，算子之间形成DAG，基于依赖划分阶段后，在阶段内形成内存迭代管道。但是MapReduce的Map和Reduce之间的交互依旧是通过硬盘来交互的。
总结：
- 编程模型上Spark占优（算子够多）
- 算子交互上，和计算行可以尽量多的内存计算而非磁盘迭代
6.4 Spark并行度
Spark的并行度：在同一时间内，有多少个task在同时运行
并行度：并行能力的设置
比如设置并行度6，其实就是要6个task并行在跑
在有了6个task并行的前提下，rdd的分区就被规划成6个分区了。
6.4.1 如何设置并行度
可以在代码中配置文件中以及提交程序的客户端参数中设置
优先级从高到低：
- 代码中
- 客户端提交参数中
- 配置文件中
- 默认（1，但是不会全部以1来跑，多数时候基于读取文件的分片数量来作为默认并行度）
全局并行度配置的参数：spark.default.parallelism
6.4.2 全局并行度-推荐
配置文件中：
conf/spark-defaults.conf 中设置
spark.default.parallelism 100
在客户端提交参数中：
bin/spark-submit --conf "spark.default.parallelism=100"
在代码中设置：
conf = SparkConf()
conf.set("spark.default.parallelism",100)
全局并行度是推荐设置，不要针对RDD改分区，可能会影响内存迭代管道的构建，或者会产生额外的shuffle
6.4.3 针对RDD的并行度设置-不推荐
只能在代码中写，算子：
- repartition算子
- coalesce算子
- partitionBy算子
6.4.4 集群中如何规划并行度
结论：设置为CPU总核心的2~10倍
比如集群可用CPU核心是100个，我们建议并行度是200~1000
确保是CPU核心的整数倍即可，最小是2倍，最大一般10倍或者更高（适量）均可
为什么要设置最少2倍？
CPU的一个核心同一时间只能干一件事。
所以，在100个核心的情况下，设置100个并行，就能让CPU100%出力。
这种设置下，如果task的压力不均衡，某个task先执行完了，就导致某个CPU核心空闲
所以，我们将Task（并行）分配的数量变多，比如100个并行，同一时间只有100个在运行，700个在等待，
但是可以确保，某个task运行完了，后续有task补上，不让cpu闲下来，最大程度利用集群的资源。
规划并行度，只看集群总CPU核数
6.5 Spark任务调度
Spark的任务，由Driver进行调度，这个工作包含：
- 逻辑DAG产生
- 分区DAG产生
- Task划分
- 将Task分配给Executor并监控其工作
如图，Spark程序的调度流程如图：
- Driver被构建出来
- 构建SparkContext（执行环境入口对象）
- 基于DAG Scheduler（DAG调度器）构建逻辑Task分配
- 基于TaskSchedule（Task调度器）将逻辑Task分配到各个Executor上干活，并监控它们。
- Worker（Executor），被TaskScheduler管理监控，听从它们的指令干活，并定期汇报进度。
1，2,3,4都是Driver的工作
5是Worker的工作
6.5.1 Drivcer内的两个组件
DAG调度器：
工作内容：将逻辑的DAG图进行处理，最终得到逻辑上的Task划分
Task调度器：
工作内容：基于DAG Scheduler的产出，来规划这些逻辑的task，应该在哪些物理的executor上运行，以及监控管理它们的运行。
6.6 拓展-Spark概念名词大全
6.6.1 Spark运行中的概念名词大全
层级关系梳理：
- 一个Spark环境可以运行多个Application
- 一个代码运行起来，会成为一个Application
- Application内部可以有多个Job
- 每个Job由一个Action产生，并且每个Job有自己的DAG执行图
- 一个Job的DAG图会基于宽窄依赖划分成不同的阶段
- 不同阶段内基于分区数量，形成多个并行的内存迭代管道
- 每一个内存迭代管道形成一个Task（DAG调度器划分将Job内划分出具体的task任务，一个Job被划分出来的task在逻辑上称之为这个job的taskset）
6.7 SparkShuffle
6.7.1MR Shuffle回顾
首先回顾MapReduce框架中Shuffle过程，整体流程图如下：
6.7.2 简介
Spark在DAG调度阶段会将一个Job划分为多个Stage，上游Stage做map工作，下游Stage做reduce工作，其本质上
还是MapReduce计算框架。Shuffle是连接map和reduce之间的桥梁，它将map的输出对应到reduce输入中，涉及
到序列化反序列化、跨节点网络IO以及磁盘读写IO等。
Spark的Shuffle分为Write和Read两个阶段，分属于两个不同的Stage，前者是Parent Stage的最后一步，后者是
Child Stage的第一步。
执行Shuffle的主体是Stage中的并发任务，这些任务分ShuffleMapTask和ResultTask两种，ShuffleMapTask要进行
Shuffle，ResultTask负责返回计算结果，一个Job中只有最后的Stage采用ResultTask，其他的均为ShuffleMapTask
。如果要按照map端和reduce端来分析的话，ShuffleMapTask可以即是map端任务，又是reduce端任务，因为
Spark中的Shuffle是可以串行的；ResultTask则只能充当reduce端任务的角色。
Spark在1.1以前的版本一直是采用Hash Shuffle的实现的方式，到1.1版本时参考Hadoop MapReduce的实现开始引
入Sort Shuffle，在1.5版本时开始Tungsten钨丝计划，引入UnSafe Shuffle优化内存及CPU的使用，在1.6中将
Tungsten统一到Sort Shuffle中，实现自我感知选择最佳Shuffle方式，到的2.0版本，Hash Shuffle已被删除，所有
Shuffle方式全部统一到Sort Shuffle一个实现中。
在Spark的中，负责shuffle过程的执行、计算和处理的组件主要就是ShuffleManager，也即shuffle管理器。ShuffleManager随着
Spark的发展有两种实现的方式，分别为HashShuffleManager和SortShuffleManager，因此spark的Shuffle有Hash Shuffle和Sort
Shuffle两种。
在Spark 1.2以前，默认的shuffle计算引擎是HashShuffleManager。该ShuffleManager而HashShuffleManager有着一个非常严重
的弊端，就是会产生大量的中间磁盘文件，进而由大量的磁盘IO操作影响了性能。
因此在Spark 1.2以后的版本中，默认的ShuffleManager改成了SortShuffleManager。SortShuffleManager相较于
HashShuffleManager来说，有了一定的改进。主要就在于，每个Task在进行shuffle操作时，虽然也会产生较多的临时磁盘文件，但
是最后会将所有的临时文件合并(merge)成一个磁盘文件，因此每个Task就只有一个磁盘文件。在下一个stage的shuffle read task拉
取自己的数据时，只要根据索引读取每个磁盘文件中的部分数据即可。
6.7.3 Sort Shuffle bypass机制
bypass运行机制的触发条件如下：
- shuffle map task数量小于spark.shuffle.sort.bypassMergeThreshold=200参数的值。
- 不是map combine聚合的shuffle算子(比如reduceByKey有map combie)。
bypass运行机制的触发条件如下：
1)shuffle map task数量小于spark.shuffle.sort.bypassMergeThreshold=200参数的值。
2)不是map combine聚合的shuffle算子(比如reduceByKey有map combie)。
- 此时task会为每个reduce端的task都创建一个临时磁盘文件，并将数据按key进行hash，然后根据key的hash值，
将key写入对应的磁盘文件之中。当然，写入磁盘文件时也是先写入内存缓冲，缓冲写满之后再溢写到磁盘文件的
。最后，同样会将所有临时磁盘文件都合并成一个磁盘文件，并创建一个单独的索引文件。
- 该过程的磁盘写机制其实跟未经优化的HashShuffleManager是一模一样的，因为都要创建数量惊人的磁盘文件，
只是在最后会做一个磁盘文件的合并而已。因此少量的最终磁盘文件，也让该机制相对未经优化的
HashShuffleManager来说，shuffle read的性能会更好。
而该机制与普通SortShuffleManager运行机制的不同在于：
第一，磁盘写机制不同;
第二，不会进行排序。也就是说，启用该机制的最大好处在于，shuffle write过程中，不需要进行数据的排序操作，
也就节省掉了这部分的性能开销
总结：
- SortShuffle也分为普通机制和bypass机制
- 普通机制在内存数据结构(默认为5M)完成排序，会产生2M个磁盘小文件。
- 而当shuffle map task数量小于spark.shuffle.sort.bypassMergeThreshold参数的值。或者算子不是聚合类的
shuffle算子(比如reduceByKey)的时候会触发SortShuffle的bypass机制，SortShuffle的bypass机制不会进行排序
，极大的提高了其性能。
6.7.4 Shuflle的配置选项
Shuffle阶段划分：
shuffle write：mapper阶段，上一个stage得到最后的结果写出
shuffle read ：reduce阶段，下一个stage拉取上一个stage进行合并
spark 的shuffle调优：主要是调整缓冲的大小，拉取次数重试重试次数与等待时间，内存比例分配，是否进行排序操作等等
spark.shuffle.file.buffer
参数说明：该参数用于设置shuffle write task的BufferedOutputStream的buffer缓冲大小（默认是32K）。将数据写到磁盘文件之前，会先写入buffer缓冲中，待缓冲写满之后，才会溢写
到磁盘。
调优建议：如果作业可用的内存资源较为充足的话，可以适当增加这个参数的大小（比如64k），从而减少shuffle write过程中溢写磁盘文件的次数，也就可以减少磁盘IO次数，进而提升性
能。在实践中发现，合理调节该参数，性能会有1%~5%的提升。
spark.reducer.maxSizeInFlight：
参数说明：该参数用于设置shuffle read task的buffer缓冲大小，而这个buffer缓冲决定了每次能够拉取多少数据。(默认48M)
调优建议：如果作业可用的内存资源较为充足的话，可以适当增加这个参数的大小（比如96m），从而减少拉取数据的次数，也就可以减少网络传输的次数，进而提升性能。在实践中发现
，合理调节该参数，性能会有1%~5%的提升。
spark.shuffle.io.maxRetries and spark.shuffle.io.retryWait：
spark.shuffle.io.maxRetries ：shuffle read task从shuffle write task所在节点拉取属于自己的数据时，如果因为网络异常导致拉取失败，是会自动进行重试的。该参数就代表了可以重试
的最大次数。（默认是3次）
spark.shuffle.io.retryWait：该参数代表了每次重试拉取数据的等待间隔。（默认为5s）
调优建议：一般的调优都是将重试次数调高，不调整时间间隔。
spark.shuffle.memoryFraction：
参数说明：该参数代表了Executor内存中，分配给shuffle read task进行聚合操作内存比例。
spark.shuffle.manager
参数说明：该参数用于设置shufflemanager的类型（默认为sort）.Spark1.5x以后有三个可选项：
Hash：spark1.x版本的默认值，HashShuffleManager
Sort：spark2.x版本的默认值，普通机制，当shuffle read task 的数量小于等于spark.shuffle.sort.bypassMergeThreshold参数，自动开启bypass 机制
spark.shuffle.sort.bypassMergeThreshold
参数说明：当ShuffleManager为SortShuffleManager时，如果shuffle read task的数量小于这个阈值（默认是200），则shuffle write过程中不会进行排序操作。
调优建议：当你使用SortShuffleManager时，如果的确不需要排序操作，那么建议将这个参数调大一些
6.8 总结
- DAG是什么有什么用？
DAG是有向无环图，用以描述任务执行流程，主要作用就是协助DAG调度器构建Task分配用以做任务管理
- 内存迭代、阶段划分？
基于DAG的宽窄依赖划分阶段，阶段内部都是窄依赖可以构建内存迭代的管道文章来源：https://www.toymoban.com/news/detail-659427.html
- DAG调度器是？
构建Task分配以做任务管理文章来源地址https://www.toymoban.com/news/detail-659427.html
到了这里，关于PySpark-核心编程的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
![林子雨 VirtualBox + Ubuntu[linux] 配置 java、hadoop、Spark[python]、pyspark快速配置流程](https://imgs.yssmx.com/Uploads/2024/04/848890-1.png)
