Spark从入门到熟悉（篇二）

本文介绍Spark的RDD编程，并进行实战演练，加强对编程的理解，实现快速入手

知识脉络

包含如下8部分内容：

创建RDD
常用Action操作
常用Transformation操作
针对PairRDD的常用操作
缓存操作
共享变量
分区操作
编程实战

创建RDD

实现方式

有如下两种方式实现：

textFile加载本地或者集群文件系统中的数据
用parallelize方法将Driver中的数据结构并行化成RDD

示例

"""
textFile: 两个参数
第一个参数: path，是必传项
第二个参数: 分区数，默认是2，可以不传。更改该参数为N再保存的话，最终输出的文件数也是N
参考说明：https://wenku.baidu.com/view/95ba5ba8e63a580216fc700abb68a98271feac80.html
"""
#从本地文件系统中加载数据
file = "./data/hello.txt"
rdd = sc.textFile(file,3)
rdd.collect()['hello world','hello spark','spark love jupyter','spark love pandas','spark love sql']

常用Action操作

介绍

Action操作将触发基于RDD依赖关系的计算。常见操作：collect、take、takeSample、first、count、reduce、foreach、countByKey、saveAsTextFile、aggregate、aggregateByKey

示例

collect: 将数据汇集到Driver,数据过大时有超内存风险take: 将前若干个数据汇集到Driver，相比collect安全takeSample: 可以随机取若干个到Driver,第一个参数设置是否放回抽样,第二个参数设置采样数量,第三个参数设置种子（可以不设置，一般用于调试，有时候不知道是程序出问题还是数据出了问题，就可以将这个参数设置为定值）
parallelize将Driver中的数据结构生成RDD,第二个参数指定分区数
rdd = sc.parallelize(range(10),5) 
sample_data = rdd.takeSample(False,10,0)
sample_data
[7, 8, 1, 5, 3, 4, 2, 0, 9, 6]first: 取第一个数据count: 查看数量reduce: 利用二元函数对数据进行规约
rdd = sc.parallelize(range(10),5) 
rdd.reduce(lambda x,y:x+y)
45foreach: 对每一个元素执行某种操作，不生成新的RDD
#累加器用法详见共享变量
rdd = sc.parallelize(range(10),5) 
accum = sc.accumulator(0)
rdd.foreach(lambda x:accum.add(x))
print(accum.value)countByKey: 对Pair RDD按key统计数量
pairRdd = sc.parallelize([(1,1),(1,4),(3,9),(2,16)]) 
pairRdd.countByKey()
defaultdict(int, {1: 2, 3: 1, 2: 1})saveAsTextFile: 保存rdd成text文件到本地
text_file = "./data/rdd.txt"
rdd = sc.parallelize(range(5))
rdd.saveAsTextFile(text_file)
#重新读入会被解析文本
rdd_loaded = sc.textFile(text_file)
rdd_loaded.collect()
['2', '3', '4', '1', '0']
#aggregate是一个Action操作
#aggregate比较复杂，先对每个分区执行一个函数，再对每个分区结果执行一个合并函数。
#例子：求元素之和以及元素个数
#三个参数，第一个参数为初始值，第二个为分区执行函数，第三个为结果合并执行函数。
rdd = sc.parallelize(range(1,21),3)
def inner_func(t,x):return((t[0]+x,t[1]+1))
def outer_func(p,q):return((p[0]+q[0],p[1]+q[1]))
rdd.aggregate((0,0),inner_func,outer_func)(210, 20)#aggregateByKey的操作和aggregate类似，但是会对每个key分别进行操作
#第一个参数为初始值，第二个参数为分区内归并函数，第三个参数为分区间归并函数
a = sc.parallelize([("a",1),("b",1),("c",2),("a",2),("b",3)],3)
b = a.aggregateByKey(0,lambda x,y:max(x,y),lambda x,y:max(x,y))
b.collect()[('b', 3), ('a', 2), ('c', 2)]

常用Transformation操作

介绍

Transformation转换操作具有懒惰执行的特性，它只指定新的RDD和其父RDD的依赖关系，只有当Action操作触发到该依赖的时候，它才被计算。常见操作如下：

map、flatMap、filter、sample、distinct；
逻辑运算（交并补）：intersection、cartesian, union, suntract
排序：sortBy
拉链方式连接：zip、zipWithIndex

示例

rdd = sc.parallelize(range(10),3)
rdd.collect()
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]map: 对每个元素进行一个映射转换
rdd.map(lambda x:x**2).collect()
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]filter: 过滤
rdd.filter(lambda x:x>5).collect()
[6, 7, 8, 9]flatMap: 将每个元素生成一个Array后压平
rdd = sc.parallelize(["hello world","hello China"])
rdd.map(lambda x:x.split(" ")).collect()
[['hello', 'world'], ['hello', 'China']]
rdd.flatMap(lambda x:x.split(" ")).collect()
['hello', 'world', 'hello', 'China']sample: 对原rdd在每个分区按照比例进行抽样,第一个参数设置是否可以重复抽样
rdd = sc.parallelize(range(10),1)
rdd.sample(False,0.5,0).collect()
[1, 4, 9]distinct: 去重
rdd = sc.parallelize([1,1,2,2,3,3,4,5])
rdd.distinct().collect()
[4, 1, 5, 2, 3]
#subtract: 找到属于前一个rdd而不属于后一个rdd的元素
a = sc.parallelize(range(10))
b = sc.parallelize(range(5,15))
a.subtract(b).collect()
[0, 1, 2, 3, 4]#union: 合并数据
a = sc.parallelize(range(5))
b = sc.parallelize(range(3,8))
a.union(b).collect()
[0, 1, 2, 3, 4, 3, 4, 5, 6, 7]#intersection: 求交集
a = sc.parallelize(range(1,6))
b = sc.parallelize(range(3,9))
a.intersection(b).collect()
[3, 4, 5]#cartesian: 笛卡尔积
boys = sc.parallelize(["LiLei","Tom"])
girls = sc.parallelize(["HanMeiMei","Lily"])
boys.cartesian(girls).collect()
[('LiLei', 'HanMeiMei'),('LiLei', 'Lily'),('Tom', 'HanMeiMei'),('Tom', 'Lily')]
#按照某种方式进行排序
#指定按照第3个元素大小进行排序
rdd = sc.parallelize([(1,2,3),(3,2,2),(4,1,1)])
rdd.sortBy(lambda x:x[2]).collect()
[(4, 1, 1), (3, 2, 2), (1, 2, 3)]
#zip: 按照拉链方式连接两个RDD，效果类似python的zip函数
#需要满足的条件: 两个RDD具有相同的分区，每个分区元素数量相同
rdd_name = sc.parallelize(["LiLei","Hanmeimei","Lily"])
rdd_age = sc.parallelize([19,18,20])
rdd_zip = rdd_name.zip(rdd_age)
print(rdd_zip.collect())
[('LiLei', 19), ('Hanmeimei', 18), ('Lily', 20)]#zipWithIndex: 将RDD和一个从0开始的递增序列按照拉链方式连接。
rdd_name =  sc.parallelize(["LiLei","Hanmeimei","Lily","Lucy","Ann","Dachui","RuHua"])
rdd_index = rdd_name.zipWithIndex()
print(rdd_index.collect())
[('LiLei', 0), ('Hanmeimei', 1), ('Lily', 2), ('Lucy', 3), ('Ann', 4), ('Dachui', 5), ('RuHua', 6)]

针对PairRDD的常用操作

介绍

PairRDD：数据为长度为2的tuple类似(k,v)结构的数据类型的RDD,其每个数据的第一个元素被当做key，第二个元素被当做value。常见操作如下：

reduceByKey、groupByKey、sortByKey
join、leftOuterJoin、rightOuterJoin
cogroup
subtractByKey
foldByKey

示例

#reduceByKey: 对相同的key对应的values应用二元归并操作
rdd = sc.parallelize([("hello",1), ("world",2),("hello",3),("world",5)])
rdd.reduceByKey(lambda x,y:x+y).collect()
[('hello', 4), ('world', 7)]#groupByKey: 将相同的key对应的values收集成一个Iterator
rdd = sc.parallelize([("hello",1),("world",2),("hello",3),("world",5)])
rdd.groupByKey().collect()
[('hello', <pyspark.resultiterable.ResultIterable at 0x119c6ae48>),('world', <pyspark.resultiterable.ResultIterable at 0x119c6a860>)]#sortByKey: 按照key排序,可以指定是否降序
rdd = sc.parallelize([("hello",1),("world",2),("China",3),("Beijing",5)])
rdd.sortByKey(False).collect()
[('world', 2), ('hello', 1), ('China', 3), ('Beijing', 5)]
#join相当于根据key进行内连接
age = sc.parallelize([("LiLei",18),("HanMeiMei",16),("Jim",20)])
gender = sc.parallelize([("LiLei","male"),("HanMeiMei","female"),("Lucy","female")])
age.join(gender).collect()
[('LiLei', (18, 'male')), ('HanMeiMei', (16, 'female'))]#leftOuterJoin相当于关系表的左连接
age = sc.parallelize([("LiLei",18),("HanMeiMei",16)])
gender = sc.parallelize([("LiLei","male"),("HanMeiMei","female"),("Lucy","female")])
age.leftOuterJoin(gender).collect()
[('LiLei', (18, 'male')), ('HanMeiMei', (16, 'female'))]#rightOuterJoin相当于关系表的右连接
age = sc.parallelize([("LiLei",18),("HanMeiMei",16),("Jim",20)])
gender = sc.parallelize([("LiLei","male"),("HanMeiMei","female")])
age.rightOuterJoin(gender).collect()#cogroup相当于对两个输入分别goupByKey然后再对结果进行groupByKey
x = sc.parallelize([("a",1),("b",2),("a",3)])
y = sc.parallelize([("a",2),("b",3),("b",5)])result = x.cogroup(y).collect()
print(result)
print(list(result[0][1][0]))[('a', (<pyspark.resultiterable.ResultIterable object at 0x119c6acc0>, <pyspark.resultiterable.ResultIterable object at 0x119c6aba8>)), ('b', (<pyspark.resultiterable.ResultIterable object at 0x119c6a978>, <pyspark.resultiterable.ResultIterable object at 0x119c6a940>))]
[1, 3]
#subtractByKey去除x中那些key也在y中的元素
x = sc.parallelize([("a",1),("b",2),("c",3)])
y = sc.parallelize([("a",2),("b",(1,2))])
x.subtractByKey(y).collect()[('c', 3)]
#foldByKey的操作和reduceByKey类似，但是要提供一个初始值
#下述示例中给的初始值为1等价于对于a有1*1*3=3,对于b有1*2*5=10
x = sc.parallelize([("a",1),("b",2),("a",3),("b",5)],1)
x.foldByKey(1,lambda x,y:x*y).collect() [('a', 3), ('b', 10)]

缓存操作

介绍

适用场景：一个rdd被多个任务用作中间量，那么对其进行cache缓存到内存中会加快计算。具体说明如下：

声明对一个rdd进行cache后，该rdd不会被立即缓存，而是等到它第一次被计算出来时才进行缓存
可以使用persist明确指定存储级别，常用的存储级别是MEMORY_ONLY和EMORY_AND_DISK。如果一个RDD后面不再用到，可以用unpersist释放缓存，unpersist是立即执行的。
缓存数据不会切断血缘依赖关系，这是因为缓存数据某些分区所在的节点有可能会有故障，例如内存溢出或者节点损坏。这时候可以根据血缘关系重新计算这个分区的数据。如果要切断血缘关系，可以用checkpoint设置检查点将某个rdd保存到磁盘中。声明对一个rdd进行checkpoint后，该rdd不会被立即保存到磁盘，而是等到它第一次被计算出来时才保存成检查点。
通常只对一些计算代价非常高昂的中间结果或者重复计算结果不可保证完全一致的情形下(如zipWithIndex算子)使用。

示例

#cache缓存到内存中，使用存储级别 MEMORY_ONLY。
#MEMORY_ONLY意味着如果内存存储不下，放弃存储其余部分，需要时重新计算。
a = sc.parallelize(range(10000),5)
a.cache()
sum_a = a.reduce(lambda x,y:x+y)
cnt_a = a.count()
mean_a = sum_a/cnt_aprint(mean_a)#persist缓存到内存或磁盘中，默认使用存储级别MEMORY_AND_DISK
#MEMORY_AND_DISK意味着如果内存存储不下，其余部分存储到磁盘中。
#persist可以指定其它存储级别，cache相当于persist(MEMORY_ONLY)
from  pyspark.storagelevel import StorageLevel
a = sc.parallelize(range(10000),5)
a.persist(StorageLevel.MEMORY_AND_DISK)
sum_a = a.reduce(lambda x,y:x+y)
cnt_a = a.count()
mean_a = sum_a/cnt_a
a.unpersist() #立即释放缓存
print(mean_a)#checkpoint 将数据设置成检查点，写入到磁盘中。
sc.setCheckpointDir("./data/checkpoint/")
rdd_students = sc.parallelize(["LiLei","Hanmeimei","LiLy","Ann"],2)
rdd_students_idx = rdd_students.zipWithIndex() 
#设置检查点后，可以避免重复计算，不会因为zipWithIndex重复计算触发不一致的问题
rdd_students_idx.checkpoint() 
rdd_students_idx.take(3)

共享变量

适用情况

当spark集群在许多节点上运行一个函数时，默认情况下会把这个函数涉及到的对象在每个节点生成一个副本。但是，有时候需要在不同节点或者节点和Driver之间共享变量。

共享变量类型

Spark提供两种类型的共享变量，广播变量和累加器。

广播变量broadcast：是不可变变量，实现在不同节点不同任务之间共享数据。广播变量在每个机器上缓存一个只读的变量，而不是为每个task生成一个副本，可以减少数据传输
累加器accumulator：主要是不同节点和Driver之间共享变量，只能实现计数或者累加功能。累加器的值只有在Driver上是可读的，在节点上不可见。

示例

#广播变量 broadcast 不可变，在所有节点可读broads = sc.broadcast(100)rdd = sc.parallelize(range(10))
print(rdd.map(lambda x:x+broads.value).collect())print(broads.value)
[100, 101, 102, 103, 104, 105, 106, 107, 108, 109]
100#累加器 只能在Driver上可读，在其它节点只能进行累加total = sc.accumulator(0)
rdd = sc.parallelize(range(10),3)rdd.foreach(lambda x:total.add(x))
total.value
45# 计算数据的平均值
rdd = sc.parallelize([1.1,2.1,3.1,4.1])
total = sc.accumulator(0)
count = sc.accumulator(0)def func(x):total.add(x)count.add(1)rdd.foreach(func)total.value/count.value
2.6

分区操作

有如下操作：

glom：将一个分区内的数据转换为一个列表作为一行。
coalesce：shuffle可选，默认为False情况下窄依赖，不能增加分区。repartition和partitionBy调用它实现。
repartition：按随机数进行shuffle，相同key不一定在同一个分区
partitionBy：按key进行shuffle，相同key放入同一个分区
mapPartitions：每次处理分区内的一批数据，适合需要分批处理数据的情况，比如将数据插入某个表，每批数据只需要开启一次数据库连接，大大减少了连接开支
mapPartitionsWithIndex：类似mapPartitions，提供了分区索引，输入参数为（i，Iterator）
foreachPartition：类似foreach，但每次提供一个Partition的一批数据
TaskContext: 获取当前分区id方法 TaskContext.get.partitionId
HashPartitioner：默认分区器，根据key的hash值进行分区，相同的key进入同一分区，效率较高，key不可为Array.
RangePartitioner：只在排序相关函数中使用，除相同的key进入同一分区，相邻的key也会进入同一分区，key必须可排序。

编程实战

求平均数

#任务：求data的平均值
data = [1,5,7,10,23,20,6,5,10,7,10]rdd_data = sc.parallelize(data)
s = rdd_data.reduce(lambda x,y:x+y+0.0)
n = rdd_data.count()
avg = s/n
print("average:",avg)

求众数

#任务：求data中出现次数最多的数，若有多个，求这些数的平均值
data =  [1,5,7,10,23,20,7,5,10,7,10]rdd_data = sc.parallelize(data)
rdd_count = rdd_data.map(lambda x:(x,1)).reduceByKey(lambda x,y:x+y)
max_count = rdd_count.map(lambda x:x[1]).reduce(lambda x,y: x if x>=y else y)
rdd_mode = rdd_count.filter(lambda x:x[1]==max_count).map(lambda x:x[0])
mode = rdd_mode.reduce(lambda x,y:x+y+0.0)/rdd_mode.count()
print("mode:",mode)

求topN

#任务：有一批学生信息表格，包括name,age,score, 找出score排名前3的学生, score相同可以任取
students = [("LiLei",18,87),("HanMeiMei",16,77),("DaChui",16,66),("Jim",18,77),("RuHua",18,50)]
n = 3rdd_students = sc.parallelize(students)
rdd_sorted = rdd_students.sortBy(lambda x:x[2],ascending = False)students_topn = rdd_sorted.take(n)
print(students_topn)

排序并返回序号

#任务：按从小到大排序并返回序号, 大小相同的序号可以不同
data = [1,7,8,5,3,18,34,9,0,12,8]rdd_data = sc.parallelize(data)
rdd_sorted = rdd_data.map(lambda x:(x,1)).sortByKey().map(lambda x:x[0])
rdd_sorted_index = rdd_sorted.zipWithIndex()print(rdd_sorted_index.collect())

二次排序

#任务：有一批学生信息表格，包括name,age,score
#首先根据学生的score从大到小排序，如果score相同，根据age从大到小students = [("LiLei",18,87),("HanMeiMei",16,77),("DaChui",16,66),("Jim",18,77),("RuHua",18,50)]
rdd_students = sc.parallelize(students)%%writefile student.py
#为了在RDD中使用自定义类，需要将类的创建代码其写入到一个文件中，否则会有序列化错误
class Student:def __init__(self,name,age,score):self.name = nameself.age = ageself.score = scoredef __gt__(self,other):if self.score > other.score:return Trueelif self.score==other.score and self.age>other.age:return Trueelse:return Falsefrom student import Studentrdd_sorted = rdd_students \.map(lambda t:Student(t[0],t[1],t[2]))\.sortBy(lambda x:x,ascending = False)\.map(lambda student:(student.name,student.age,student.score))#参考方案：此处巧妙地对score和age进行编码来表达其排序优先级关系，除非age超过100000，以下逻辑无错误。
#rdd_sorted = rdd_students.sortBy(lambda x:100000*x[2]+x[1],ascending=False)rdd_sorted.collect()[('LiLei', 18, 87),('Jim', 18, 77),('HanMeiMei', 16, 77),('DaChui', 16, 66),('RuHua', 18, 50)]

连接操作

#任务：已知班级信息表和成绩表，找出班级平均分在75分以上的班级
#班级信息表包括class,name,成绩表包括name,scoreclasses = [("class1","LiLei"), ("class1","HanMeiMei"),("class2","DaChui"),("class2","RuHua")]
scores = [("LiLei",76),("HanMeiMei",80),("DaChui",70),("RuHua",60)]rdd_classes = sc.parallelize(classes).map(lambda x:(x[1],x[0]))
rdd_scores = sc.parallelize(scores)
rdd_join = rdd_scores.join(rdd_classes).map(lambda t:(t[1][1],t[1][0]))def average(iterator):data = list(iterator)s = 0.0for x in data:s = s + xreturn s/len(data)rdd_result = rdd_join.groupByKey().map(lambda t:(t[0],average(t[1]))).filter(lambda t:t[1]>75)
print(rdd_result.collect())

分组求众数

#任务：有一批学生信息表格，包括class和age。求每个班级学生年龄的众数。students = [("class1",15),("class1",15),("class2",16),("class2",16),("class1",17),("class2",19)]def mode(arr):dict_cnt = {}for x in arr:dict_cnt[x] = dict_cnt.get(x,0)+1max_cnt = max(dict_cnt.values())most_values = [k for k,v in dict_cnt.items() if v==max_cnt]s = 0.0for x in most_values:s = s + xreturn s/len(most_values)rdd_students = sc.parallelize(students)
rdd_classes = rdd_students.aggregateByKey([],lambda arr,x:arr+[x],lambda arr1,arr2:arr1+arr2)
rdd_mode = rdd_classes.map(lambda t:(t[0],mode(t[1])))print(rdd_mode.collect())[('class1', 15.0), ('class2', 16.0)]

结尾

亲爱的读者朋友：感谢您在繁忙中驻足阅读本期内容！您的到来是对我们最大的支持❤️

正如古语所言："当局者迷，旁观者清"。您独到的见解与客观评价，恰似一盏明灯💡，能帮助我们照亮内容盲区，让未来的创作更加贴近您的需求。

若此文给您带来启发或收获，不妨通过以下方式为彼此搭建一座桥梁： ✨ 点击右上角【点赞】图标，让好内容被更多人看见 ✨ 滑动屏幕【收藏】本篇，便于随时查阅回味 ✨ 在评论区留下您的真知灼见，让我们共同碰撞思维的火花

我始终秉持匠心精神，以键盘为犁铧深耕知识沃土💻，用每一次敲击传递专业价值，不断优化内容呈现形式，力求为您打造沉浸式的阅读盛宴📚。

有任何疑问或建议？评论区就是我们的连心桥！您的每一条留言我都将认真研读，并在24小时内回复解答📝。

愿我们携手同行，在知识的雨林中茁壮成长🌳，共享思想绽放的甘甜果实。下期相遇时，期待看到您智慧的评论与闪亮的点赞身影✨！

万分感谢🙏🙏您的点赞👍👍、收藏⭐🌟、评论💬🗯️、关注❤️💚～

自我介绍：一线互联网大厂资深算法研发（工作6年+），4年以上招聘面试官经验（一二面面试官，面试候选人400+），深谙岗位专业知识、技能雷达图，已累计辅导15+求职者顺利入职大中型互联网公司。熟练掌握大模型、NLP、搜索、推荐、数据挖掘算法和优化，提供面试辅导、专业知识入门到进阶辅导等定制化需求等服务，助力您顺利完成学习和求职之旅（有需要者可私信联系）

友友们，自己的知乎账号为“快乐星球”，定期更新技术文章，敬请关注！