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、搜索、推薦、數據挖掘算法和優化，提供面試輔導、專業知識入門到進階輔導等定制化需求等服務，助力您順利完成學習和求職之旅（有需要者可私信聯系）

友友們，自己的知乎賬號為“快樂星球”，定期更新技術文章，敬請關注！???