全棧工程師開發手冊 （作者：欒鵬）

獲取SparkContext

python語法
1. 獲取sparkSession：  se = SparkSession.builder.config(conf = SparkConf()).getOrCreate()
2. 獲取sparkContext：  sc = se.sparkContext
3. 獲取sqlContext：    sq = SparkSession.builder.getOrCreate()
4. 獲取DataFrame：     df = sqlContext.createDataFrame(userRows)

資料格式

1. [[u'3', u'5'], [u'4', u'6'], [u'4', u'5'], [u'4', u'2']]   拆分或擷取的原始資料，  可以通過 map 中的 x[0], x[1] 來獲取對應列的資料
　　可以通過 map 來轉換為key-value 資料格式  例如：  df3 = df2.map(lambda x: (x[0], x[1]))    //python語法

2. key-value 資料格式
　　[(u'3', u'5'), (u'4', u'6'), (u'4', u'5'), (u'4', u'2')]   中每一個（） 表示一組資料，  第一個表示key    第二個表示value
3）PipelinedRDD  型別表示   key-value形式資料
 

RDD

RDD是Spark中的抽象資料結構型別，任何資料在Spark中都被表示為RDD。從程式設計的角度來看，RDD可以簡單看成是一個數組。和普通陣列的區別是，RDD中的資料是分割槽儲存的，這樣不同分割槽的資料就可以分佈在不同的機器上，同時可以被並行處理。因此，Spark應用程式所做的無非是把需要處理的資料轉換為RDD，然後對RDD進行一系列的變換和操作從而得到結果。

建立RDD：
（1）、通過讀取本地或者hdfs上的檔案建立RDD

//scala語法
path = "hdfs://master:9000/examples/examples/src/main/resources/people.txt"  
rdd1 = sc.textFile(path,2)  

path = "file:///usr/local/spark/spark-1.6.0-bin-hadoop2.6/README.md"  //local file  
rdd1 = sc.textFile(path,2)  
 

（2）通過並行化的方式建立RDD. 其實就是通過我們自己取模擬資料

//scala語法
# list轉RDD
sc.parallelize([1,2,3,4,5], 3)  #意思是將陣列中的元素轉換為RDD，並且儲存在3個分割槽上[1]、[2,3]、[4,5]。如果是4個分割槽：[1]、[2]、[3]、[4,5]

val str=Array("you jump","I jump")
val list = Array(1,2,3,4,5,6)
val listadd = sc.parallelize(list) 可以看到返回值就是RDD，當然可以呼叫RDD中的函式，比如reduce運算元等
listadd.reduce(_+_)


# rdd轉list
list = RDD.collect()

只要搞懂了spark的函式們，你就成功了一大半。

spark的函式主要分兩類，Transformations和Actions。Transformations為一些資料轉換類函式，actions為一些行動類函式：

轉換：轉換的返回值是一個新的RDD集合，而不是單個值。呼叫一個變換方法，不會有任何求值計算，它只獲取一個RDD作為引數，然後返回一個新的RDD。

行動：行動操作計算並返回一個新的值。當在一個RDD物件上呼叫行動函式時，會在這一時刻計算全部的資料處理查詢並返回結果值。

transformation 會針對已有的RDD建立一個新的RDD，而action則主要對RDD進行最後的操作。transformation只是記錄了對RDD的操作，並不會觸發spark程式的執行，只有當transform之後接著一個action操作，那麼所有的transformation才會執行。比如

//scala語法
val file=sc.textFile("hdfs://hadoop1:9000/hello.txt").flatMap(line => line.split("\t"))
回車之後並沒有觸發spark的執行，因為flatMap等屬於transformation操作
等到file.collect()後會看到spark的執行，collect是action操作

下面介紹spark常用的Transformations, Actions函式：

Transformations

map(func [, preservesPartitioning=False]) — 返回一個新的分散式資料集，這個資料集中的每個元素都是經過func函式處理過的。

//python語法
>>> data = [1,2,3,4,5]
>>> distData = sc.parallelize(data).map(lambda x: x+1).collect()
#結果：[2,3,4,5,6]

filter(func) — 返回一個新的資料集，這個資料集中的元素是通過func函式篩選後返回為true的元素（簡單的說就是，對資料集中的每個元素進行篩選，如果符合條件則返回true，不符合返回false，最後將返回為true的元素組成新的資料集返回）。

//python語法
>>> rdd = sc.parallelize(data).filter(lambda x:x%2==0).collect()
#結果：[2, 4]

flatMap(func [, preservesPartitioning=False]) — 類似於map(func)， 但是不同的是map對每個元素處理完後返回與原資料集相同元素數量的資料集，而flatMap返回的元素數不一定和原資料集相同。每個輸入元素可能被影射為0個或多個輸出元素。

//python語法
#### for flatMap()
>>> rdd = sc.parallelize([2,3,4])
>>> sorted(rdd.flatMap(lambda x: range(1,x)).collect())
#結果：[1, 1, 1, 2, 2, 3]
>>> sorted(rdd.flatMap(lambda x:[(x,x), (x,x)]).collect())
#結果：[(2, 2), (2, 2), (3, 3), (3, 3), (4, 4), (4, 4)]
 
#### for map()
>>> rdd = sc.parallelize([2,3,4])
>>> sorted(rdd.map(lambda x: range(1,x)).collect())
#結果：[[1], [1, 2], [1, 2, 3]]
>>> sorted(rdd.map(lambda x:[(x,x), (x,x)]).collect())
#結果：[[(2, 2), (2, 2)], [(3, 3), (3, 3)], [(4, 4), (4, 4)]]

mapPartitions(func [, preservesPartitioning=False]) —mapPartitions是map的一個變種。map的輸入函式是應用於RDD中每個元素，而mapPartitions的輸入函式是應用於每個分割槽，也就是把每個分割槽中的內容作為整體來處理的。

//python語法
>>> rdd = sc.parallelize([1,2,3,4,5], 3)
>>> def f(iterator): yield sum(iterator)
>>> rdd.mapPartitions(f).collect()
#結果：[1,5,9]

mapPartitionsWithIndex(func [, preservesPartitioning=False]) —類似於mapPartitions, 只是包含兩個引數.第一個引數為分割槽的索引，第二個引數為每個分割槽內的元素迭代器。輸出是一個包含元素列表的迭代器。

//python語法
>>> rdd = sc.parallelize([1,2,3,4,5], 3)
>>> def f(splitIndex, iterator): yield splitIndex
>>> rdd.mapPartitionsWithIndex(f).collect()
#結果：[0,1,2]   #三個分割槽的索引

reduceByKey(func [, numPartitions=None, partitionFunc=<function portable_hash at 0x7fa664f3cb90>]) — reduceByKey就是對元素為kv對的RDD中Key相同的元素的value進行reduce，因此，key相同的多個元素的值被reduce為一個值，然後與原RDD中的key組成一個新的kv對。等價於pandas中的groupby

//python語法
>>> from operator import add
>>> rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
>>> sorted(rdd.reduceByKey(add).collect())
>>> #或者 sorted(rdd.reduceByKey(lambda a,b:a+b).collect())
#結果：[('a', 2), ('b', 1)]

sortByKey([ascending=True, numPartitions=None, keyfunc=<function <lambda> at 0x7fa665048c80>]) 返回排序後的資料集。該函式就是對kv對的RDD資料進行排序，keyfunc是對key進行處理的函式，如非需要，不用管。 也就是說先按keyfunc對key進行處理，然後按照key進行排序。

//python語法
>>> tmp = [('a', 1), ('b', 2), ('1', 3), ('D', 4)]
>>> sc.parallelize(tmp).sortByKey(True, 1).collect()
#結果： [('1', 3), ('D', 4), ('a', 1), ('b', 2)]
>>> sc.parallelize(tmp).sortByKey(True, 2, keyfunc=lambda k:k.lower()).collect()
#結果：[('1', 3), ('a', 1), ('b', 2), ('D', 4)]
#注意，比較兩個結果可看出，keyfunc對鍵的處理只是在資料處理的過程中起作用，不能真正的去改變鍵名

join(otherDataset [, numPartitions=None]) — join就是對元素為kv對的RDD中key相同的value收集到一起組成(v1,v2)，然後與原RDD中的key組合成一個新的kv對，返回。即RDD1中的元素與RDD2中元素具有相同key值的嘗試合併。

//python語法
>>> x = sc.parallelize([("a", 1), ("b", 4)])
>>> y = sc.parallelize([("a", 2), ("a", 3)])
>>> sorted(x.join(y).collect())
#結果：[('a', (1, 2)), ('a', (1, 3))]

cartesian(otherDataset) — 返回一個笛卡爾積的資料集，這個資料集是通過計算兩個RDDs得到的。即RDD1中的元素與RDD2中元素任意兩個元素嘗試合併。

//python語法
>>> x = sc.parallelize([1,2,3])
>>> y = sc.parallelize([4,5])
>>> x.cartesian(y).collect()
#結果：[(1, 4), (1, 5), (2, 4), (2, 5), (3, 4), (3, 5)]

RDD 刪除 重複 元素

intRdd.distinct()     #去重

隨機將一個 RDD 通過指定比例 分為 2 個RDD

sRdd = stringRdd.randomSplit([0.4,0.6])    將 stringRdd 以4:6 分為2個 RDD，  獲取其中一個 RDD 的方法為： sRdd[0]

RDD 中 groupBy 分組計算

//python語法
gRdd = intRdd.groupBy(lambda x: x<2)   #將會分為2組，  訪問第一組：  print(sorted(gRdd[0][1])),  訪問第二組：print(sorted(gRdd[1][1]))，因為gRdd[0][0]和gRdd[1][0]表示預設生成的組名

分組並且取別名：  gRdd = intRdd.groupBy(lambda x: "a" if(x < 2) else "b"),   
（1）獲取第一組資訊： print(gRdd[0][0], sorted(gRdd[0][1]))
（2） 獲取第二組資訊： print(gRdd[1][0], sorted(gRdd[1][1]))   其中，   前半部分 gRdd[1][0] 表示獲取別名 a

Action (這裡只講支援python的)

reduce(func) — reduce將RDD中元素兩兩傳遞給輸入函式，同時產生一個新的值，新產生的值與RDD中下一個元素再被傳遞給輸入函式直到最後只有一個值為止。

>>> from operator import add
>>> sc.parallelize([1,2,3,4,5]).reduce(add)
# 結果：15

collect() — 返回RDD中的資料，以list形式。

>>> sc.parallelize([1,2,3,4,5]).collect()
#結果：[1,2,3,4,5]

count() — 返回RDD中的元素個數。

>>> sc.parallelize([1,2,3,4,5]).count
#結果：5

first() — 返回RDD中的第一個元素。

>>> sc.parallelize([1,2,3,4,5]).first()
#結果：1

take(n) — 返回RDD中前n個元素。

>>> sc.parallelize([1,2,3,4,5]).take(2)
#結果：[1,2]

takeOrdered(n [, key=None]) — 返回RDD中前n個元素，但是是升序(預設)排列後的前n個元素，或者是通過key函式指定後的RDD（這個key我也沒理解透，後面在做詳解）。等價於sql語句裡面 order limit

>>> sc.parallelize([9,7,3,2,6,4]).takeOrdered(3)
#結果：[2,3,4]
>>> sc.parallelize([9,7,3,2,6,4]).takeOrdered(3, key=lambda x:-x)
#結果：[9,7,6]

1） intRdd1.stats() 統計 intRdd1， 結果為：(count: 5, mean: 5.0, stdev: 2.82842712475, max: 9, min: 1) mean表示平均值， stdev 表示標準差
2）intRdd3.min() 最新值，
3）intRdd3.max() 最大值
4）intRdd3.stdev() 標準差
5）intRdd3.count() 資料條數
6）intRdd3.sum() 求和
7）intRdd3.mean() 平均值
　　
saveAsTextFile(path [, compressionCodecClass=None]) — 該函式將RDD儲存到檔案系統裡面，並且將其轉換為文字行的檔案中的每個元素呼叫 tostring 方法。path - 保存於檔案系統的路徑，compressionCodecClass壓縮編碼型別，如gzip

>>> tempFile = NamedTemporaryFile(delete=True)
>>> tempFile.close()
>>> sc.parallelize(range(10)).saveAsTextFile(tempFile.name)
>>> from fileinput import input
>>> from glob import glob
>>> ''.join(sorted(input(glob(tempFile.name + "/part-0000*"))))
'0\n1\n2\n3\n4\n5\n6\n7\n8\n9\n'

在儲存檔案的時候，空白行也是允許存在的。

>>> tempFile2 = NamedTemporaryFile(delete=True)
>>> tempFile2.close()
>>> sc.parallelize(['', 'foo', '', 'bar', '']).saveAsTextFile(tempFile2.name)
>>> ''.join(sorted(input(glob(tempFile2.name + "/part-0000*"))))
'\n\n\nbar\nfoo\n'

使用壓縮編碼

>>> tempFile3 = NamedTemporaryFile(delete=True)
>>> tempFile3.close()
>>> codec = "org.apache.hadoop.io.compress.GzipCodec"
>>> sc.parallelize(['foo', 'bar']).saveAsTextFile(tempFile3.name, codec)
>>> from fileinput import input, hook_compressed
>>> result = sorted(input(glob(tempFile3.name + "/part*.gz"), openhook=hook_compressed))
>>> b''.join(result).decode('utf-8')
u'bar\nfoo\n'

countByKey() — 返回一個字典（key,count），該函式操作資料集為kv形式的資料，用於統計RDD中擁有相同key的元素個數。

>>> defdict = sc.parallelize([("a",1), ("b",1), ("a", 1)]).countByKey()
>>> defdict
#結果：defaultdict(<type 'int'>, {'a': 2, 'b': 1})
>>> defdict.items()
#結果：[('a', 2), ('b', 1)]

countByValue() — 返回一個字典（value,count），該函式操作一個list資料集，用於統計RDD中擁有相同value的元素個數。

>>> sc.parallelize([1,2,3,1,2,5,3,2,3,2]).countByValue().items()
#結果：[(1, 2), (2, 4), (3, 3), (5, 1)]

foreach(func) — 執行函式func來處理RDD中的每個元素，這個函式常被用來updating an Accumulator或者與外部儲存系統的互動。

>>> def f(x): print(x)
>>> sc.parallelize([1, 2, 3, 4, 5]).foreach(f)
#note: 列印是隨機的，並不是一定按1,2,3,4,5的順序列印

RDD key-value 基本轉換運算

1）kvRdd1 = sc.parallelize([(1, 4),(2, 5),(3, 6),(4, 7)])  建立RDD key-value 源資料
　　結果為：  [(1, 4), (2, 5), (3, 6), (4, 7)]
2）kvRdd1.keys()    獲取全部 key 的值
3)kvRdd1.values()   獲取全部 values 的值
4)kvRdd1.filter(lambda keyValue: keyValue[0] > 2)  過濾 key > 2 的資料
5)kvRdd1.filter(lambda keyValue: keyValue[1] >5)   過濾 value > 5 的資料
6)kvRdd1.mapValues(lambda x: x*x)  對每一條 value 進行運算
7)kvRdd1.sortByKey()  按照 key 從小到大 進行排序
8)kvRdd1.sortByKey(ascending=False)  按照 key 從大到小進行排序
9)kvRdd3.reduceByKey(lambda x, y:x+y)   將 key 相同的鍵的值合併相加

多個 RDD key-value 的轉換運算

1） join
　　intK1 = sc.parallelize([(1,5),(2,6),(3,7),(4,8),(5,9)])
　　intK2 = sc.parallelize([(3,30),(2,20),(6,60)])
　　intK1.join(intK2)   join結果為：
　　[(2, (6, 20)), (3, (7, 30))]
2）leftJoin
　　intK1.leftOuterJoin(intK2).collect()   leftJoin結果為：
　　[(2, (6, 20)), (4, (8, None)), (1, (5, None)), (3, (7, 30)), (5, (9, None))]
3）rightJoin　　　　rightJoin 結果為：
　　intK1.rithtOuterJoin(intK2).collect()
　　[(2, (6, 20)), (6, (None, 60)), (3, (7, 30))]
4）subtractByKey    從 intK1 中刪除 與 intK2 相同 key-value
　　intK1.subtractByKey(intK2)    結果為：
　　[(4, 8), (1, 5), (5, 9)]　　

Spark Shell的使用

Spark’s shell 提供了一種學習API的簡單方法, 也是一個強大的資料分析工具.它可以通過Scala(它是一種執行在Java虛擬機器上並且能夠使用Java庫的程式語言 )或者Python來操作. 在Spark 目錄下執行以下指令碼:

./bin/spark-shell (Scala)
./bin/pyspark (Python)

Spark-shell有兩種使用方式：

1：直接Spark-shell
會啟動一個SparkSubmit程序來模擬Spark執行環境，是一個單機版的。

2：Spark-shell --master Spark：//hadoop1:7077,hadoop2:7077,hadoop3:7077 --total-executor-cores 5 --executor-memory 5G

指定任務提交的叢集路徑在哪裡。這就需要提前啟動一個真實的Standalone叢集。

可以指定多個master的地址，用逗號隔開。

如果沒有指定–total-executor-cores 5 --executor-memory 5G，那麼就會使用叢集中所有可用的資源，每一個worker中都會啟動executor。

spark提交指令碼命令（spark-submit使用及說明）：

Python指令碼中需要在開頭匯入spark相關模組，呼叫時使用spark-submit提交

簡單的demo

本地檔案/lp/demo/data.txt

a a b d e te w qw
d r as r a  b d a x
d s a v h f
as das 
das g h r

python檔案/lp/demo/wordcount.py

"""用Python編寫的一個簡單Spark應用"""

import sys
from operator import add

from pyspark import SparkContext


if __name__ == "__main__":
    
    sc= SparkContext.getOrCreate( )
    
    path = 'file:///lp/demo/data.txt'
    # 第一種方法
    rdd1 = sc.textFile(path)
    counts = rdd1.flatMap(lambda x: x.split(' ')).map(lambda x: (x, 1)).reduceByKey(add)
    print(type(counts))
    output = counts.collect()
    for (word, count) in output:
        print("%s: %i" % (word, count))

    # 第二中方法
    result = rdd1.flatMap(lambda x: x.split(' ')).countByValue().items()
    print(result)

    # 第三種方法
    result = rdd1.flatMap(lambda x: x.split(' ')).map(lambda x: (x, 1)).countByKey().items()
    print(result)

本機執行的模式執行python檔案

[[email protected] spark-2.3.0-bin-hadoop2.7]# ./bin/spark-submit /lp/demo/wordcount.py

在spark叢集上執行python檔案（Standalone模式）

[[email protected] spark-2.3.0-bin-hadoop2.7]# ./bin/spark-submit --master spark://localhost:7077 /lp/demo/wordcount.py

直接使用python直譯器執行。

python /lp/demo/wordcount.py

python直譯器執行帶有spark程式碼的python指令碼

如果使用python解釋執行帶有spark程式碼的python指令碼會報錯。

ImportError: No module named pyspark
ImportError: No module named py4j.java_gateway

缺少pyspark和py4j這兩個模組，這兩個包在Spark的安裝目錄裡，需要在環境變數裡定義PYTHONPATH，編輯~/.bashrc或者/etc/profile檔案均可

vi ~/.bashrc # 或者 sudo vi /etc/profile
# 新增下面這一行
export PYTHONPATH=$SPARK_HOME/python/:$SPARK_HOME/python/lib/py4j-0.10.1-src.zip:$PYTHONPATH
# 使其生效
source ~/.bashrc # 或者 sudo source /etc/profile

然後關閉終端，重新開啟，用python執行即可。重起後執行

python mypy.py