Python3:Python+spark程式設計實戰 總結
阿新 • • 發佈:2020-09-14
0、前提
0.1 配置
有關spark
說明:
spark2.0.1的版本還不支援Python3.6
安裝注意版本
anaconda4.2(python3.5)
可參考:
windows上配置 Python+spark開發環境
一、例項分析
1.1 資料 student.txt
1.2 程式碼
#studentExample 例子 練習 def map_func(x): s = x.split() return (s[0], [int(s[1]),int(s[2]),int(s[3])]) #返回為(key,vaklue)格式,其中key:x[0],value:x[1]且為有三個元素的列表 #return (s[0],[int(s[1],s[2],s[3])]) #注意此用法不合法 def has100(x): for y in x: if(y == 100): #把x、y理解為 x軸、y軸 return True return False def allis0(x): if(type(x)==list and sum(x) == 0): #型別為list且總分為0 者為true;其中type(x)==list :判斷型別是否相同 return True return False def subMax(x,y): m = [x[1][i] if(x[1][i] > y[1][i]) else y[1][i] for i in range(3)] return('Maximum subject score', m) def sumSub(x,y): n = [x[1][i]+y[1][i] for i in range(3)] #或者 n = ([x[1][0]+y[1][0],x[1][1]+y[1][0],x[1][2]+y[1][2]]) return('Total subject score', n) def sumPer(x): return (x[0],sum(x[1])) #停止之前的SparkContext,不然重新執行或者建立工作會失敗;另外,只有 sc.stop()也可以,但是首次執行會有誤 try: sc.stop() except: pass from pyspark import SparkContext #匯入模組 sc=SparkContext(appName='Student') #命名 lines=sc.textFile("student.txt").map(lambda x:map_func(x)).cache() #匯入資料且保持在記憶體中,其中cache():資料保持在記憶體中 count=lines.count() #對RDD中的資料個數進行計數;其中,RDD一行為一個數據集 #RDD'轉換'運算 (篩選 關鍵字filter) whohas100 = lines.filter(lambda x: has100(x[1])).collect() #注意:處理的是value列表,也就是x[1] whois0 = lines.filter(lambda x: allis0(x[1])).collect() sumScore = lines.map(lambda x: (x[0],sum(x[1]))).collect() #‘動作’運算 maxScore = max(sumScore,key=lambda x: x[1]) #總分最高者 minScore = min(sumScore,key=lambda x: x[1]) #總分最低者 sumSubScore = lines.reduce(lambda x,y: sumSub(x,y)) avgScore = [x/count for x in sumSubScore[1]]#單科成績平均值 #RDD key-value‘轉換’運算 subM = lines.reduce(lambda x,y: subMax(x,y)) redByK = lines.reduceByKey(lambda x,y: [x[i]+y[i] for i in range(3)]).collect() #合併key相同的value值x[0]+y[0],x[1]+y[1],x[2]+y[2] #RDD'轉換'運算 sumPerSore = lines.map(lambda x: sumPer(x)).collect() #每個人的總分 #sumSore = lines.map(lambda x: (x[0],sum(x[1]))).collect() sorted = lines.sortBy(lambda x: sum(x[1])) #總成績低到高的學生成績排序 sortedWithRank = sorted.zipWithIndex().collect()#按總分排序 first3 = sorted.takeOrdered(3,key=lambda x:-sum(x[1])) #總分前三者 #限定以空格的形式輸出到檔案中 first3RDD = sc.parallelize(first3)\ .map(lambda x:str(x[0])+' '+str(x[1][0])+' '+str(x[1][1])+' '+str(x[1][2])).saveAsTextFile("result")
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
#print(lines.collect()) print("資料集個數(行):",count) print("單科滿分者:",whohas100) print("單科零分者:",whois0) print("單科最高分者:",subM) print("單科總分:",sumSubScore) print("合併名字相同的分數:",redByK) print("總分/(人)",sumPerSore) print("最高總分者:",maxScore) print("最低總分者:",minScore) print("每科平均成績:",avgScore) print("總分倒序:",sortedWithRank) print("總分前三者:",first3) print(first3RDD) sc.stop()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
1.3 結果展示
資料集個數(行): 7 單科滿分者: [('li', [100, 54, 0]), ('li', [100, 54, 0])] 單科零分者: [('yanf', [0, 0, 0])] 單科最高分者: ('Maximum subject score', [100, 90, 100]) 單科總分: ('Total subject score', [485, 438, 280]) 合併名字相同的分數: [('li', [200, 108, 0]), ('zhang', [180, 180, 200]), ('yang', [85, 90, 30]), ('wang', [20, 60, 50]), ('yanf', [0, 0, 0])] 總分/(人) [('yang', 205), ('wang', 130), ('zhang', 280), ('zhang', 280), ('li', 154), ('li', 154), ('yanf', 0)] 最高總分者: ('zhang', 280) 最低總分者: ('yanf', 0) 每科平均成績: [69.28571428571429, 62.57142857142857, 40.0] 總分倒序: [(('yanf', [0, 0, 0]), 0), (('wang', [20, 60, 50]), 1), (('li', [100, 54, 0]), 2), (('li', [100, 54, 0]), 3), (('yang', [85, 90, 30]), 4), (('zhang', [90, 90, 100]), 5), (('zhang', [90, 90, 100]), 6)] 總分前三者: [('zhang', [90, 90, 100]), ('zhang', [90, 90, 100]), ('yang', [85, 90, 30])] None
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
二、程式碼解析
2.1函式解析
2.1.1 collect()
RDD的特性
在進行基本RDD“轉換”運算時不會立即執行,結果不會顯示在顯示屏中,collect()是一個“動作”運算,會立刻執行,顯示結果。
2.1.2 reduce()
說明
reduce()函式會對引數序列中的元素進行累積。
語法
reduce(function, iterable[, initializer])
引數
- function – 函式,有兩個引數
- iterable – 可迭代物件
- initializer – 可選,初始引數
例項
說明:Python3的內建函式移除了reduce函式,reduce函式放在functools模組
In [24]:
#r = reduce(lambda x, y: x+y, [4,4,5,5]) # 使用 lambda 匿名函式
from functools import reduce
def add(x, y) : # 兩數相加
return x + y
reduce(add, [1,2,3,4,5])
Out[24]:
15
In [25]:
reduce(lambda x, y: x+y, [1,2,3,4,5]) # 使用 lambda 匿名函式
Out[25]:
15
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
2.1.3 type()
語法
class type(name, bases, dict)
引數
- name – 類的名稱。
- bases – 基類的元組。
- dict – 字典,類內定義的名稱空間變數。
返回值
一個引數返回物件型別, 三個引數,返回新的型別物件。
例項
# 一個引數例項
In [1]:
type(1)
Out[1]:
int
In [2]:
type([2])
Out[2]:
list
In [3]:
type({3:'three'})
Out[3]:
dict
In [5]:
x = 5
type(x) == list #判斷x的型別是否為list
Out[5]:
False
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
# 三個引數例項
class y(object):
z = 5
x = type('y',(object,),dict(z=5))
print(x)
<class '__main__.y'> #產生一個新的型別
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
三、問題分析
An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 1 in stage 2.0 failed 1 times, most recent failure: Lost task 1.0 in stage 2.0 (TID 5, localhost, executor driver): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
- 1
- 2
解析
1、檢查拼寫是否有誤
2、檢查縮排是否合規
3、檢查()是否一一配對
四、例項 小練
4.1 資料 user_small
1441900799.728000 1441900802.452000 8618245698655 0134730038729312 2 1 1 IPHONE_5 17999 20693 10.67.23.157 111.13.34.100 6 58986 80 GET mmsns.qpic.cn /mmsns/PdibpV1sFDHdaOTqNXb8VGSNicyYpOVa9R7icxSr4BkwbsSyzJbBTmE5Zz5aZichejbkKuia7twzraqk/150?tp=webp&length=1136&width=640 weixin.qq.com/?version=369229843&uin=2925174340&nettype=0&scene=moment WeChat/6.2.0.19 CFNetwork/711.3.18 Darwin/14.0.0 200 59 image/webp 7504 706 8212 7 1827
1441900750.023000 1441900754.063000 8613836044032 0136210021269713 2 1 1 IPHONE_5 17752 25632 10.67.21.71 117.144.242.26 6 52941 80 POST short.weixin.qq.com http://short.weixin.qq.com/cgi-bin/micromsg-bin/tenpay - MicroMessenger Client - - - - 715 0 7 1827
1441900755.480472 1441900756.762000 8618246899077 0131830068670612 2 1 1 IPHONE_4S 17875 61433 10.67.43.51 120.192.84.86 6 58684 31271 GET i.gtimg.cn http://i.gtimg.cn/qqshow/admindata/comdata/vip_emoji_aio_ios_new_config/xydata.json - QQ/5.7.0.469 CFNetwork/672.0.8 Darwin/14.0.0 304 83 x-json - 0 0 18 1041
1441900754.860000 1441900755.480472 8618246899077 0131830068670612 2 1 1 IPHONE_4S 17875 61433 10.67.43.51 120.192.84.86 6 58684 31271 GET i.gtimg.cn http://i.gtimg.cn/club/item/avatar/zip/0/i0/all.zip - QQ/5.7.0.469 CFNetwork/672.0.8 Darwin/14.0.0 404 210 text/html 85 487 411 18 1041
1441900753.786000 1441900755.726000 8618246195634 9900026543899411 2 1 1 IPHONE_4S 17783 19302 10.67.29.55 111.40.194.207 6 49412 80 GET sb.symcd.com /MFYwVKADAgEAME0wSzBJMAkGBSsOAwIaBQAEFDmvGLQcAh85EJZW%2FcbTWO90hYuZBBROQ8gddu83U3pP8lhvlPM44tW93wIQd9jUM82by0%2FVy957MNapGQ%3D%3D - securityd (unknown version) CFNetwork/672.0.2 Darwin/14.0.0 - - - - 522 0 18 1041
1441900761.308739 1441900761.408000 8615045213668 0127590050857822 2 1 1 IPHONE_4 17772 50621 10.67.63.219 183.232.95.61 6 49337 80 POST szminorshort.weixin.qq.com http://szminorshort.weixin.qq.com/cgi-bin/micromsg-bin/rtkvreport - MicroMessenger Client - - - - 500 16 7 1827
1441900696.427624 1441900761.308739 8615045213668 0127590050857822 2 1 1 IPHONE_4 17772 50621 10.67.63.219 183.232.95.61 6 49337 80 POST szminorshort.weixin.qq.com http://szminorshort.weixin.qq.com/cgi-bin/micromsg-bin/rtkvreport - MicroMessenger Client - - - - 500 16 7 1827
1441900693.219000 1441900696.427624 8615045213668 0127590050857822 2 1 1 IPHONE_4 17772 50621 10.67.63.219 183.232.95.61 6 49337 80 POST szminorshort.weixin.qq.com http://szminorshort.weixin.qq.com/cgi-bin/micromsg-bin/rtkvreport - MicroMessenger Client - - - - 502 16 7 1827
1441900750.845345 1441900753.537000 8618246195634 9900026543899411 2 1 1 IPHONE_4S 17783 19302 10.67.29.55 117.135.169.124 6 49411 80 GET b227.photo.store.qq.com /psb?/V12jlwSP30SPej/VE1V5LlXFMzHeg5gTzpyuCueaEVEGV*0X6BbSyJZRhs!/b/dCWGUIc.HQAA&ek=1&kp=1&pt=0&bo=yAD6AAAAAAABBxI!&t=5 v1_iph_sq_5.6.0_1_app_a-4-2 QQ/5.6.0.438 CFNetwork/672.0.2 Darwin/14.0.0 - - - - 792 0 18 1041
1441900748.094000 1441900750.845345 8618246195634 9900026543899411 2 1 1 IPHONE_4S 17783 19302 10.67.29.55 117.135.169.124 6 49411 80 GET b227.photo.store.qq.com /psb?/V12jlwSP30SPej/VE1V5LlXFMzHeg5gTzpyuCueaEVEGV*0X6BbSyJZRhs!/b/dCWGUIc.HQAA&ek=1&kp=1&pt=0&bo=yAD6AAAAAAABBxI!&t=5 v1_iph_sq_5.6.0_1_app_a-4-2 QQ/5.6.0.438 CFNetwork/672.0.2 Darwin/14.0.0 - - - - 792 0 18 1041
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
4.2 使用者上網記錄統計(一行為一條記錄).(使用者:第3列)
#test 1_1 使用者上網記錄統計
sc.stop()
from pyspark import SparkContext
sc = SparkContext(appName='test1')
rdd = sc.textFile('user_small')\
.map(lambda x:x.split('\t'))\
.map(lambda x:(x[3],1))\
.reduceByKey(lambda x,y:x+y)\
.map(lambda x:str(x[0])+' '+str(x[0][1])).collect()
#.saveAsTextFile('text1_1') #限定為空格鍵輸出到檔案
print(rdd)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
['0127590050857822 1', '9900026543899411 9', '0131830068670612 1', '0136210021269713 1', '0134730038729312 1']
- 1
4.2使用者流量統計。分別統計上行流量及下行流量並將結果各列以空格鍵隔開輸出到檔案。(使用者:第3列;上行流量:第25列;下行流量:第26列)
#test 1_2 統計使用者上網 分別為上、下行流量
def map_func(x):
s = x.split('\t')
return (s[2],[int(s[24]),int(s[25])])#返回為(key,vaklue)格式,其中key:x[0],value:x[1]且為有三個元素的列表
#return (s[0],[int(s[1],s[2],s[3])]) #注意此用法不合法
try:
sc.stop() #停止之前的SparkContext,不然重新執行或者建立工作會失敗
except:
pass
from pyspark import SparkContext
sc=SparkContext(appName='test')
lines=sc.textFile("user_small").map(lambda x:map_func(x)).cache()
redByK = lines.reduceByKey(lambda x,y: (x[0]+y[0],x[1]+y[1]))
sum_flow = redByK.map(lambda x:str(x[0])+' '+str(x[1][0])+' '+str(x[1][1]))\
.saveAsTextFile('text1_2')
sc.stop()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
4.3 統計使用者總流量
#test 1_2 統計使用者上網 總流量
try:
sc.stop() #停止之前的SparkContext,不然重新執行或者建立工作會失敗
except:
pass
from pyspark import SparkContext
sc = SparkContext(appName='test1')
rdd = sc.textFile('user_small')\
.map(lambda x:x.split('\t'))\
.map(lambda x:(x[2],int(x[24])+int(x[25])))\
.reduceByKey(lambda x,y:x+y)\
.map(lambda x:str(x[0])+' '+str(x[1])).collect()
print(rdd)
sc.stop()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
['8618246899077 898', '8615045213668 1550', '8618245698655 8918', '8613836044032 715', '8618246195634 2106']
- 1
4.4、微信APP流量統計。(微信APP特徵MicroMessenger,位於第20列,統計對應的下行流量值——第26列的數值。)
#test 1_3
sc.stop()
from pyspark import SparkContext
sc = SparkContext(appName='test1')
rdd = sc.textFile('user_small')\
.map(lambda x:x.split('\t'))\
.map(lambda x:(x[19],int(x[25])))\
.filter(lambda x: 'WeChat' or 'MicroMessenger' in x[1])#篩選\
.reduceByKey(lambda x,y:x+y)\
.map(lambda x:str(x[0])+' '+str(x[1])).collect()
print(rdd)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
['securityd (un