在開始之前，我先介紹一下，RDD是什么？

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

　　創建RDD：

1	>>> sc.parallelize([1,2,3,4,5],?3)??#意思是將數組中的元素轉換為RDD，并且存儲在3個分區上[1]、[2,3]、[4,5]。如果是4個分區：[1]、[2]、[3]、[4,5]

?　　上面這種是數組創建，也可以從文件系統或者HDFS中的文件創建出來，后面會講到。

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

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

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

行動：行動操作計算并返回一個新的值。當在一個RDD對象上調用行動函數時，會在這一時刻計算全部的數據處理查詢并返回結果值。

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

Transformations

map(func [,?preservesPartitioning=False])? --- 返回一個新的分布式數據集，這個數據集中的每個元素都是經過func函數處理過的。

123	>>> 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的元素組成新的數據集返回）。

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

flatMap(func [,?preservesPartitioning=False])? --- 類似于map(func)， 但是不同的是map對每個元素處理完后返回與原數據集相同元素數量的數據集，而flatMap返回的元素數不一定和原數據集相同。each input item can be mapped to 0 or more output items (so?funcshould return a Seq rather than a single item)

#### for flatMap() >>> rdd = sc.parallelize([2,3,4]) >>> sorted(rdd.flatMap(lambda x: range(1,x)).collect()) #結果：[1, 1, 1, 2, 2, 3] func=lambda x:range(1,x) print"上面的這個輸出結果可能不太好理解，可以根據下面的代碼的輸出結果來理解，其實就是所有的結果被打散后拼接起來了" print(func(1)) print(func(2)) print(func(3)) print(func(4)) print"-"*50 >>> 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.flatMap(lambda x: range(1,x)).collect()) #結果：[[1], [1, 2], [1, 2, 3]] >>> sorted(rdd.flatMap(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的輸入函數是應用于每個分區，也就是把每個分區中的內容作為整體來處理的。

?

1234	>>> 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])? ---Similar to?mapPartitions, but takes two parameters. The first parameter is the index of the partition and the second is an iterator through all the items within this partition. The output is an iterator containing the list of items after applying whatever transformation the function encodes.

1234	>>> 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對。

12345

>>>?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)]

aggregateByKey(zeroValue)(seqOp, combOp [,?numPartitions=None])? ---

sortByKey([ascending=True,?numPartitions=None,?keyfunc=<function <lambda> at 0x7fa665048c80>])? --- 返回排序后的數據集。該函數就是隊kv對的RDD數據進行排序，keyfunc是對key進行處理的函數，如非需要，不用管。

123456

>>> 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()#上面的這個k.lower()的意思是會返回大寫字母的小寫形式,所以最終變成了a,b,d一起比較大小 #其中的2的意思是分成幾個區的意思 #結果：[('1', 3), ('a', 1), ('b', 2), ('D', 4)]#注意，比較兩個結果可看出，keyfunc對鍵的處理只是在數據處理的過程中起作用，不能真正的去改變鍵名

oin(otherDataset [,?numPartitions=None])? --- join就是對元素為kv對的RDD中key相同的value收集到一起組成(v1,v2)，然后與原RDD中的key組合成一個新的kv對，返回。

1234	>>> x?=?sc.parallelize([("a",?1), ("b",?4)])>>> y?=?sc.parallelize([("a",?2), ("a",?5)])>>>?sorted(x.join(y).collect())#結果：[('a', (1, 2)), ('a', (1, 5))]

cartesian(otherDataset)? --- 返回一個笛卡爾積的數據集，這個數據集是通過計算兩個RDDs得到的。

1234	>>> 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)]

上面這個其實就是排列組合.

Action?(這里只講支持python的，java和scala的后面用到了在做詳解，當然支持python就一定支持java和scala)

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

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

collect()? --- 返回RDD中的數據，以list形式。

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

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

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

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

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

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

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

takeOrdered(n [,?key=None])? --- 返回RDD中前n個元素，但是是升序(默認)排列后的前n個元素，或者是通過key函數指定后的RDD（這個key我也沒理解透，后面在做詳解）

1234	>>> 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]

saveAsTextFile(path [,?compressionCodecClass=None])? --- 該函數將RDD保存到文件系統里面，并且將其轉換為文本行的文件中的每個元素調用 tostring 方法。

parameters:? path - 保存于文件系統的路徑

　　　　　　　compressionCodecClass - (None by default) string i.e. “org.apache.hadoop.io.compress.GzipCodec”

import tempfile from tempfile import TemporaryFile,NamedTemporaryFiletempFile = NamedTemporaryFile(delete=True) tempFile.close() print"tempFile.name=",tempFile.namesc.parallelize(range(10)).saveAsTextFile(tempFile.name) from fileinput import input from glob import glob#這是用來查找文件路徑名的 print ''.join(sorted(input(glob(tempFile.name + "/part-0000*")))) #以上代碼的最終結果就是把0-9保存在文件:/tmp/tmpwocaZg/part-00000中

?Empty lines are tolerated when saving to text files:

12345

>>> 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'

?Using compressionCodecClass:

12345678

>>> 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的元素個數。

12345

>>> 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的元素個數。

12	>>> 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或者與外部存儲系統的交互。

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

總結

以上是生活随笔為你收集整理的spark 常用函数介绍（python）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。