- 運(yùn)行spark程序首先需要聲明SparkSession 創(chuàng)建一個(gè)自己的集群(下面使用單機(jī)演示)
聲明SparkSession
很多書上就只有spark，sc等字符澳腹，如果不查資料很難知道這些從哪里來
from pyspark.sql import SparkSession
# 創(chuàng)建一個(gè)spark 回話孽尽，appName任意取
spark = SparkSession.builder.appName('learnspark').getOrCreate()
# sc用于讀取外部數(shù)據(jù)耿焊，spark功能非常強(qiáng)大涝焙，可以讀取文件,s3,hdfs等等數(shù)據(jù)
sc = spark.sparkContext
- 讀取文件的方法很多墓贿，這里只介紹讀取文件(我在工作中一般用s3讀取储笑，具體原理有待研究)
# 創(chuàng)建RDD的方法有兩種,第一種使用外部文件
# 這里最后一個(gè)參數(shù)4表示數(shù)據(jù)集被劃分分區(qū)個(gè)數(shù)甜熔，這里劃分了4個(gè)分區(qū)
data = sc.textFile("/FileStore/tables/VS14MORT_txt-b42fa.gz", 4)
# 第二種方式，使用paralleliz集合
# ***一般來說突倍，建議把每個(gè)數(shù)據(jù)集分為2-4個(gè)分區(qū)
data2 = sc.parallelize([['Amber', 23],['Bob',32], ['Lucy', 18], ['Amber', 12], ['Kent', 40], ['Julia', 25]], 4)
- 轉(zhuǎn)換腔稀，這里的 ‘轉(zhuǎn)換’ 是種惰性操作，實(shí)際上類似于把指令記錄下來羽历，到真正需要用的在執(zhí)行
1. map()函數(shù)：類似于python中的map焊虏，對rdd中每一行進(jìn)行操作
data2.map(lambda x: x[0]).take(10)
[out]:['Amber', 'Bob', 'Lucy', 'Amber', 'Kent', 'Julia']
# 注意，純python方法會(huì)降低程序的運(yùn)行速度秕磷，盡量使用spark內(nèi)置的功能
2. filter()函數(shù)：類似于python中的filter诵闭，篩選滿足條件的行
data2 = data2.filter(lambda x: x[1]>20)
data2.collect()
[out]: [['Amber', 23], ['Bob', 32], ['Kent', 40], ['Julia', 25]]
3. flatMap()函數(shù)：將結(jié)果作為一個(gè)list輸出
data2 = data2.flatMap(lambda x: (x[0],x[0]))
data2.collect()
[out]: ['Amber', 'Amber', 'Bob', 'Bob', 'Kent', 'Kent', 'Julia', 'Julia']
# 必須是tulpe或者list,不然一個(gè)字符一個(gè)字符輸出
data2 = data2.flatMap(lambda x: x[0])
data2.collect()
[out]: ['A', 'm', 'b', 'e', 'r','B','o',..] # 太多了只截取前一部分
# 如果要正常輸出應(yīng)考慮lambda x: (x[0], )
4. distinct()函數(shù)：類似于set()
data2 = data2.map(lambda x:x[0]).distinct()
data2.collect()
[out]:['Amber', 'Julia', 'Bob', 'Lucy', 'Kent']
# 注意這是一個(gè)高開銷的方法,一般和map一起用
5. leftOuterJoin()函數(shù)：左連接
rdd1 = sc.parallelize([('a', 1), ('b', 4), ('c',10)])
rdd2 = sc.parallelize([('a', 4), ('a', 1), ('b', '6'), ('d', 15)])
rdd3 = rdd1.leftOuterJoin(rdd2) # rdd2去匹配rdd1
rdd3.collect()
[out]: [('c', (10, None)), ('b', (4, '6')), ('a', (1, 4)), ('a', (1, 1))]
#這是一個(gè)高開銷的方法
6. join()函數(shù)：相當(dāng)于 innerjoin兩個(gè)都有才合并
rdd3 = rdd1.join(rdd2)
rdd3.collect()
[out]:[('b', (4, '6')), ('a', (1, 4)), ('a', (1, 1))]
7. intersection()函數(shù)：這里貌似rdd中元素必須完全一樣才輸出，有待研究...
rdd1 = sc.parallelize([('a', 2), ('b', 4), ('c',10)])
rdd2 = sc.parallelize([('a', 4), ('a', 1), ('b', '6'), ('d', 15)])
rdd3 = rdd1.intersection(rdd2)
print(rdd3.collect()) # 返回空澎嚣，沒有完全一樣的
[out]:[]
rdd1 = sc.parallelize([('a', 1), ('b', 4), ('c',10)])
rdd2 = sc.parallelize([('a', 4), ('a', 1), ('b', '6'), ('d', 15)])
rdd3 = rdd1.intersection(rdd2)
print(rdd3.collect())
[out]:[('a', 1)]
8. repartition()函數(shù)：重新分區(qū)疏尿，這個(gè)功能很重要
在公司遇到過數(shù)據(jù)量太大, 集群默認(rèn)分區(qū)數(shù)量默認(rèn)值太小的情況，這里就需要重新分區(qū)
rdd1 = sc.parallelize([('a', 2), ('b', 4), ('c',10)])
print(rdd1.glom().collect(), len(rdd1.glom().collect())) # 最開始放在8分區(qū)里面易桃，其中5個(gè)沒有數(shù)據(jù)
[out]: [[], [], [('a', 2)], [], [], [('b', 4)], [], [('c', 10)]] 8
rdd1 = rdd1.repartition(3)
print(rdd1.glom().collect(), len(rdd1.glom().collect()))# 現(xiàn)在存放在4個(gè)分區(qū)里面褥琐，其中1個(gè)沒數(shù)據(jù)
[out]: [[], [('a', 2), ('b', 4), ('c', 10)], []] 3
rdd1 = rdd1.repartition(2)
print(rdd1.glom().collect(), len(rdd1.glom().collect()))# 現(xiàn)在存放在2個(gè)分區(qū)里面，其中一個(gè)分區(qū)存2個(gè)數(shù)據(jù)
[out]: [[('b', 4)], [('a', 2), ('c', 10)]] 2
- 操作晤郑，‘操作’的函數(shù)敌呈，會(huì)把‘轉(zhuǎn)換’的付諸實(shí)施
1. take()函數(shù)：取前n條返回記錄
rdd1 = sc.parallelize([('a', 2), ('b', 4), ('c',10)])
rdd1 = rdd1.map(lambda x: (x[0], x[1]+1))
rdd1.take(1)
[out]: [('a', 3)]
2. takeSample()函數(shù)：隨機(jī)取得n個(gè)記錄
rdd1 = sc.parallelize([('a', 2), ('b', 4), ('c',10)])
rdd1.takeSample(True, 5) # 第一個(gè)參數(shù)是否為有放回抽樣嚼鹉，第二個(gè)參數(shù)n條記錄，第三個(gè)參數(shù)seed
[out]: [('a', 2), ('c', 10), ('b', 4), ('a', 2), ('a', 2)]
3. reduce()函數(shù)：類似于python reduce
rdd1 = sc.parallelize([('a', 2), ('b', 4), ('c',10)])
rdd1.map(lambda x:x[1]).reduce(lambda x,y:x+y) # 這里不需要collect
[out]: 16
# 注意驱富，如果是/則要看分區(qū)
rdd1 = rdd1.repartition(2)
rdd1.map(lambda x:x[1]).reduce(lambda x,y:x/y) # 2/(4/10) 如果再改分區(qū)可能會(huì) (2/4)/10
[out]: 5.0
4. count()函數(shù)：返回rdd元素總數(shù)
rdd1.count()
[out]: 3
5. saveAsTextFile()函數(shù)：保存數(shù)據(jù)到文本
data_key = sc.parallelize([('a', 4),('b', 3),('c', 2),('a', 8),('d', 2),('b', 1),('d', 3)],4)
data_key.saveAsTextFile('/FileStore/tables/tmp.txt')
# 這里[('a', 4),('b', 3),('c', 2),('a', 8),('d', 2),('b', 1),('d', 3)]
# 會(huì)變?yōu)閟tr([('a', 4),('b', 3),('c', 2),('a', 8),('d', 2),('b', 1),('d', 3)]) 字符串
# 讀取時(shí)需要轉(zhuǎn)換
def parseInput(row):
row_split = eval(row) # 字符串需要用eval還原
return row_split
data_key = sc.textFile('/FileStore/tables/tmp.txt').map(parseInput)
data_key.collect()
[out]: [('a', 4),('b', 3),('c', 2),('a', 8),('d', 2),('b', 1),('d', 3)]
6. foreach()函數(shù)：和map功能類似锚赤，但是是for循環(huán)一個(gè)接一個(gè)做
def f(x): # 這個(gè)函數(shù)在jupyter中不輸出，需要在終端中才能輸出
print(x)
data_key.foreach(f)
7. collect()函數(shù)：rdd轉(zhuǎn)化為python list, 運(yùn)行時(shí)需考慮單機(jī)的內(nèi)存
data_key.collect()
[out]: [('a', 4), ('b', 3), ('c', 2), ('a', 8), ('d', 2), ('b', 1), ('d', 3)]
