本節書摘來自華章出版社《spark大資料分析：核心概念、技術及實踐》一書中的第3章，第3.5節，作者[美] 穆罕默德·古勒（mohammed guller），更多章節内容可以通路雲栖社群“華章計算機”公衆号檢視。
3.5 api
應用可以通過使用spark提供的庫獲得spark叢集計算的能力。這些庫都是用scala編寫的。但是spark提供了各種語言的api。在本書編寫之際，spark api提供了如下語言的支援：scala、java、python和r。可以使用上面的任何語言來開發spark應用。也有其他語言（比如clojure）的非官方支援。
spark api主要由兩個抽象部件sparkcontext和彈性分布式資料集（rdd）構成。應用程式通過這兩個部件和spark進行互動。應用程式可以連接配接到spark叢集并使用相關資源。接下來會介紹這兩個抽象部件，然後詳細介紹rdd。
3.5.1 sparkcontext
sparkcontext是一個在spark庫中定義的類。它是spark庫的入口點。它表示與spark叢集的一個連接配接。使用spark api建立的其他一些重要對象都依賴于它。
每個spark應用程式都必須建立一個sparkcontext類執行個體。目前，每個spark應用程式隻能擁有一個激活的sparkcontext類執行個體。如果要建立一個新的執行個體，那麼在此之前必須讓目前激活的類執行個體失活。
sparkcontext有多個構造函數。最簡單的一個不需要任何參數。一個sparkcontext類執行個體可以用如下代碼建立。
在這種情況下，sparkcontext的配置資訊都從系統屬性中擷取，比如spark master的位址、應用名稱等。也可以建立一個sparkconf類執行個體，然後把它作為sparkcontext的參數進而設定配置資訊。sparkconf 是spark庫中定義的一個類。通過這種方式可以像下面這樣設定各種spark配置資訊。
sparkconf為設定諸如spark master這樣的常用配置資訊都提供了對應的顯式方法。此外，它還提供了一個通用的方法用于設定配置資訊，它使用鍵-值對進行設定。sparkcontext和sparkconf可以使用的參數将在第4章進行詳細介紹。
在本章接下來的例子中會繼續使用上面建立的變量sc。
3.5.2 rdd
彈性分布式資料集（rdd）表示一個關于分區資料元素的集合，可以在其上進行并行操作。它是spark的主要資料抽象概念。它是spark庫中定義的一個抽象類。
從概念上看，除了可以用于表示分布式資料集和支援惰性操作的特性外，rdd類似于spark的集合。惰性操作将在本章稍後部分詳細介紹。
下面分别簡要描述rdd的特點。
不可變性
rdd是一種不可變的資料結構。一旦建立，它将不可以在原地修改。基本上，一個修改rdd的操作都會傳回一個新的rdd。
分片
rdd表示的是一組資料的分區。這些分區分布在多個叢集節點上。然而，當spark在單個節點運作時，所有的分區資料都會在目前節點上。
spark存儲rdd的分區和資料集實體分區之間關系的映射關系。rdd是各個分布式資料源之中資料的一個抽象，它通常表示分布在多個叢集節點上的分區資料。比如hdfs将資料分片或分塊分散存儲在叢集中。預設情況下，一個rdd分區對應一個hdfs檔案分片。其他的分布式資料源（比如cassandra）同樣也将資料分片分散存儲在叢集多個節點上。然而，一個rdd對應多個cassandra分片。
容錯性
rdd為可容錯的。rdd代表了分散在叢集中多個節點的資料，但是任何一個節點都有可能出故障。誠如之前所說的，一個節點出故障的可能性和叢集節點數量成正比。叢集越大，在任何一個節點它出故障的可能性就越高。
rdd會自動處理節點出故障的情況。當一個節點出故障時，該節點上存儲的資料将無法被通路。此時，spark會在其他節點上重建丢失的rdd分區資料。spark存儲每一個rdd的血統資訊。通過這些血統資訊，spark可以恢複rdd的部分資訊，當節點出故障的時候，它甚至可以恢複整個rdd。
接口
需要着重指出的是，rdd是一個處理資料的接口。在spark庫中它定義為一個抽象類。rdd為多種資料源提供了一個處理資料的統一接口，包括hdfs、hbase、cassandra等。這個接口同樣可以用于處理存儲于多個節點記憶體中的資料。
spark為不同資料源提供了各自具體的實作類，比如hadooprdd、parallelcollection-rdd、jdbcrdd和cassandrardd。它們都支援基礎的rdd接口。
強類型
rdd類有一個參數用于表示類型，這使得rdd可以表示不同類型的資料。rdd可以表示同一類型資料的分布式集合，包括integer、long、float、string或者應用開發者自己定義的類型。而且，一個應用總會使用某種類型的rdd，包括integer、long、float、double、string或自定義類型。
駐留在記憶體中
之前已經提及了spark的記憶體叢集計算特性。rdd類提供一套支援記憶體計算的api。spark允許rdd在記憶體中緩存或長期駐留。就像之前所說的，對一個緩存在記憶體中的rdd進行操作比操作沒緩存的rdd要快很多。
3.5.3 建立rdd
由于rdd是一個抽象類，是以無法直接建立一個rdd的類執行個體。sparkcontext類提供了一個工廠方法用來建立rdd實作類的類執行個體。rdd也可以通過由其他rdd執行轉換操作得到。就像之前所說的，rdd是不可變的。任何一個對rdd的修改操作都将傳回一個代表修改後資料的新rdd。
本節總結了幾種建立rdd的常見方法。在下面的示例代碼中，sc是一個sparkcontext的類執行個體。之前的章節已經介紹了怎麼建立它。
parallelize
這個方法用于從本地scala集合建立rdd執行個體。它會對scala集合中的資料重新分區、重新分布，然後傳回一個代表這些資料的rdd。這個方法很少用在生産上，但是使用它有助于學習spark。
textfile
textfile方法用于從文本檔案建立rdd執行個體。它可以從多種來源讀取資料，包括單個檔案、本地同一目錄下的多個檔案、hdfs、amazon s3，或其他hadoop支援的存儲系統。這個方法傳回一個rdd，這個rdd代表的資料集每個元素都是一個字元串，每一個字元串代表輸入檔案中的一行。
上面的代碼表示從存儲于hdfs上的一個檔案或者目錄建立rdd執行個體。
textfile方法也可以讀取壓縮檔案中的資料。而且，它的參數中可以存在通配符，用于從一個目錄中讀取多個檔案。下面是一個例子。
textfile的第二個參數是一個可選參數，它用于指定分區的個數。預設情況下，spark為每一個檔案分塊建立一個分區。可以設定成一個更大的數字進而提高并行化程度，但是設定成一個小于檔案分塊數的數字是不可以的。
wholetextfiles
這個方法讀取目錄下的所有文本檔案，然後傳回一個由鍵值型rdd。傳回rdd中的每一個鍵值對對應一個檔案。鍵為檔案路徑，對應的值為該檔案的内容。這個方法可以從多種來源讀取資料，包括本地檔案系統、hdfs、amazon s3，或者其他hadoop支援的存儲系統。
sequencefile
sequencefile方法從sequencefile檔案中擷取鍵值對資料，這些sequencefile檔案可以存儲于本地檔案系統、hdfs或者其他hadoop支援的存儲系統。這個方法傳回一個鍵值對型rdd執行個體。當使用這個方法的時候，不僅需要提供檔案名，還需要提供檔案中資料鍵和值各自的類型。
3.5.4 rdd操作
spark應用使用rdd類或其繼承類中定義的方法來處理資料。這些方法也稱為操作。既然scala中可以把一個方法當成操作符使用，那麼rdd中的方法有時也稱為操作符。
spark的美好之處就在于同樣一個rdd方法既可以處理幾位元組的資料也可以處理pb級的資料。而且spark應用可以使用同樣的方法去處理資料，無論它是存儲于本地還是存儲于一個分布式存儲系統。這樣的靈活性使得開發者可以在單機上開發、調試、測試spark應用，然後不用改動任何代碼就可以将它部署到一個大叢集上。
rdd操作可以歸為兩類：轉換和行動。轉換将會建立一個新的rdd執行個體。行動則會将結果傳回給驅動程式。
轉換
轉換指的是在原rdd執行個體上進行計算，而後建立一個新的rdd執行個體。本節将介紹一些常見的轉換操作。
從概念上看，rdd轉換操作的類似于scala集合上的方法。主要的差別在于scala集合方法操作的資料是在單機記憶體中的，而rdd的轉換操作可以處理分布在叢集各個節點上的資料。另外一個重要的差別是，rdd轉換操作是惰性的，而scala集合方法不是。本章餘下部分會詳細介紹這些内容。
map
map方法是一個高階方法，它把一個函數作為它的參數，并把這個函數作用在原rdd的每個元素上，進而建立一個新rdd執行個體。這個作為參數的函數擁有一個參數并傳回一個值。
filter
filter方法是一個高階方法，它把一個布爾函數作為它的參數，并把這個函數作用在原rdd的每個元素上，進而建立一個新rdd執行個體。一個布爾函數隻有一個參數作為輸入，傳回true或false。filter方法傳回一個新的rdd執行個體，這個rdd執行個體代表的資料集由布爾函數傳回true的元素構成。是以，新rdd執行個體代表的資料集是原rdd的子集。
flatmap
flatmap方法是一個高階方法，它把一個函數作為它的參數，這個函數處理原rdd中每個元素傳回一個序列。扁平化這個序列的集合得到一個資料集，flatmap方法傳回的rdd就代表這個資料集。
mappartitions
mappartitions是一個高階方法，它使你可以以分區的粒度來處理資料。相比于一次處理一個元素，mappartitions一次處理處理一個分區，每個分區被當成一個疊代器。mappartitions方法的函數參數把疊代器作為輸入，傳回另外一個疊代器作為輸出。map-partitions将自定義函數參數作用于每一個分區上，進而傳回一個新rdd執行個體。
union
union方法把一個rdd執行個體作為輸入，傳回一個新rdd執行個體，這個新rdd執行個體的資料集是原rdd和輸入rdd的合集。
intersection
intersection方法把一個rdd執行個體作為輸入，傳回一個新rdd執行個體，這個新rdd執行個體代表的資料集是原rdd和輸入rdd的交集。
這是另外一個例子。
subtract
subtract方法把一個rdd執行個體作為輸入，傳回一個新rdd執行個體，這個新rdd執行個體代表的資料集由那些存在于原rdd執行個體中但不在輸入rdd執行個體中的元素構成。
distinct
rdd執行個體上的distinct方法傳回一個新rdd執行個體，這個新rdd執行個體的資料集由原rdd的資料集去重後得到。
cartesian
cartesian方法把一個rdd執行個體作為輸入，傳回一個新rdd執行個體，這個新rdd執行個體的資料集由原rdd和輸入rdd的所有元素的笛卡兒積構成。傳回的rdd執行個體的每一個元素都是一個有序二進制組，每一個有序二進制組的第一個元素來自原rdd，第二個元素來自輸入rdd。元素的個數等于原rdd的元素個數乘以輸入rdd的元素個數。
這個方法類似于sql中的join操作。
zip
zip方法把一個rdd執行個體作為輸入，傳回一個新rdd執行個體，這個新rdd執行個體的每一個元素是一個二進制組，二進制組的第一個元素來自原rdd，第二個元素來自輸入rdd。和cartesian方法不同的是，zip方法傳回的rdd的元素個數于原rdd的元素個數。原rdd的元素個數和輸入rdd的相同。進一步地說，原rdd和輸入rdd不僅有相同的分區數，每個分區還有相同的元素個數。
zipwithindex
zipwithindex方法傳回一個新rdd執行個體，這個新rdd執行個體的每個元素都是由原rdd元素及其下标構成的二進制組。
groupby
groupby是一個高階方法，它将原rdd中的元素按照使用者定義的标準分組進而組成一個rdd。它把一個函數作為它的參數，這個函數為原rdd中的每一個元素生成一個鍵。groupby把這個函數作用在原rdd的每一個元素上，然後傳回一個由二進制組構成的新rdd執行個體，每個二進制組的第一個元素是函數生成的鍵，第二個元素是對應這個鍵的所有原rdd元素的集合。其中，鍵和原rdd元素的對應關系由那個作為參數的函數決定。
需要注意的是，groupby是一個費時操作，因為它可能需要對資料做shuffle操作。
假設有一個csv檔案，檔案的内容為公司客戶的姓名、年齡、性别和郵編。下面的示例代碼示範了按照郵編将客戶分組。
keyby
keyby方法與groupby方法相類似。它是一個高階方法，把一個函數作為參數，這個函數為原rdd中的每一個元素生成一個鍵。keyby方法把這個函數作用在原rdd的每一個元素上，然後傳回一個由二進制組構成的新rdd執行個體，每個二進制組的第一個元素是函數生成的鍵，第二個元素是對應這個鍵的原rdd元素。其中，鍵和原rdd元素的對應關系由那個作為參數的函數決定。傳回的rdd執行個體的元素個數和原rdd的相同。
groupby和keyby的差別在于傳回rdd執行個體的元素上。雖然都是二進制組，但是
groupby傳回的二進制組中的第二個元素是一個集合，而keyby的是單個值。
sortby
sortby是一個高階方法，它将原rdd中的元素進行排序後組成一個新的rdd執行個體傳回。它擁有兩個參數。第一個參數是一個函數，這個函數将為原rdd的每一個元素生成一個鍵。第二個參數用來指明是升序還是降序排列。
下面是另一個示例。
pipe
pipe方法可以讓你建立子程序來運作一段外部程式，然後捕獲它的輸出作為字元串，用這些字元串構成rdd執行個體傳回。
randomsplit
randomsplit方法将原rdd分解成一個rdd數組。它的參數是分解的權重。
coalesce
coalesce方法用于減少rdd的分區數量。它把分區數作為參數，傳回分區數等于這個參數的rdd執行個體。
使用coalesce方法時需要小心，因為減少了rdd的分區數也就意味着降低了spark的并行能力。它通常用于合并小分區。舉例來說，在執行filter操作之後，rdd可能會有很多小分區。在這種情況下，減少分區數能提升性能。
repartition
repartition方法把一個整數作為參數，傳回分區數等于這個參數的rdd執行個體。它有助于提高spark的并行能力。它會重新分布資料，是以它是一個耗時操作。
coalesce和repartition方法看起來一樣，但是前者用于減少rdd中的分區，後者用于增加rdd中的分區。
sample
sample方法傳回原rdd資料集的一個抽樣子集。它擁有三個參數。第一個參數指定是有放回抽樣還是無放回抽樣。第二個參數指定抽樣比例。第三個參數是可選的，指定抽樣的随機數種子。
鍵值對型rdd的轉換
除了上面介紹的rdd轉換之外，針對鍵值對型rdd還支援其他的一些轉換。下面将介紹隻能作用于鍵值對型rdd的常用轉換操作。
keys
keys方法傳回隻由原rdd中的鍵構成的rdd。
values
values方法傳回隻由原rdd中的值構成的rdd。
mapvalues
mapvalues是一個高階方法，它把一個函數作為它的參數，并把這個函數作用在原rdd的每個值上。它傳回一個由鍵值對構成的rdd。它和map方法類似，不同點在于它把作為參數的函數作用在原rdd的值上，是以原rdd的鍵都沒有變。傳回的rdd和原rdd擁有相同的鍵。
join
join方法把一個鍵值對型rdd作為參數輸入，而後在原rdd和輸入rdd上做内連接配接操作。它傳回一個由二進制組構成的rdd。二進制組的第一個元素是原rdd和輸入rdd都有的鍵，第二個元素是一個元組，這個元組由原rdd和輸入rdd中鍵對應的值構成。
leftouterjoin
leftouterjoin方法把一個鍵值對型rdd作為參數輸入，而後在原rdd和輸入rdd之間做左連接配接操作。它傳回一個由鍵值對構成的rdd。鍵值對的第一個元素是原rdd中的鍵，第二個元素是一個元組，這個元組由原rdd中鍵對應的值和輸入rdd中的可選值構成。可選值用option類型表示。
rightouterjoin
rightouterjoin方法把一個鍵值對型rdd作為參數輸入，而後在原rdd和輸入rdd之間做右連接配接操作。它傳回一個由鍵值對構成的rdd。鍵值對的第一個元素是輸入rdd中的鍵，第二個元素是一個元組，這個元組由原rdd中的可選值和輸入rdd中鍵對應的值構成。可選值用option類型表示。
fullouterjoin
fullouterjoin方法把一個鍵值對型rdd作為參數輸入，而後在原rdd和輸入rdd之間做全連接配接操作。它傳回一個由鍵值對構成的rdd。
samplebykey
samplebykey通過在鍵上抽樣傳回原rdd的一個子集。它把對每個鍵的抽樣比例作為輸入參數，傳回原rdd的一個抽樣。
subtractbykey
subtractbykey方法把一個鍵值對型rdd作為輸入參數，傳回一個鍵值對rdd，這個鍵值對rdd的鍵都是隻存在原rdd中但是不存在于輸入rdd中。
groupbykey
groupbykey方法傳回一個由二進制組構成的rdd，二進制組的第一個元素是原rdd的鍵，第二個元素是一個集合，集合由該鍵對應的所有值構成。它類似于上面介紹過的group-by方法。二者的差別在于groupby是一個高階方法，它的參數是一個函數，這個函數為原rdd的每一個元素生成一個鍵。groupbykey方法作用于rdd的每一個鍵值對上，故不需要一個生成鍵的函數作為輸入參數。
應當盡量避免使用groupbykey。它是一個耗時操作，因為它可能會對資料進行shuffle操作。在大多數情況下，都有不使用groupbykey的更好的替代方案。
reducebykey
reducebykey是一個高階方法，它把一個滿足結合律的二進制操作符當作輸入參數。它把這個操作符作用于有相同鍵的值上。
一個二進制操作符把兩個值當作輸入參數，傳回一個值。一個滿足結合律的二進制操作符傳回同樣的結果，但是它不關心操作數的分組情況。
reducebykey方法可以用于對同一鍵對應的值進行彙總操作。比如它可以用于對同一鍵對應的值進行求和，求乘積，求最小值，求最大值。
對于基于鍵的彙總操作、合并操作，reducebykey比groupbykey更合适。
操作
操作指的是那些傳回值給驅動程式的rdd方法。本節介紹一些rdd中常用的操作。
collect
collect方法傳回一個數組，這個數組由原rdd中的元素構成。在使用這個方法的時候需要小心，因為它把在worker節點的資料移給了驅動程式。如果操作一個有大資料集的rdd，它有可能會導緻驅動程式崩潰。
count
count方法傳回原rdd中元素的個數。
countbyvalue
countbyvalue方法傳回原rdd中每個元素的個數。它傳回是一個map類執行個體，其中，鍵為元素的值，值為該元素的個數。
first
first方法傳回原rdd中的第一個元素。
max
max方法傳回rdd中最大的元素。
min
min方法傳回rdd中最小的元素。
take
take方法的輸入參數為一個整數n，它傳回一個由原rdd中前n個元素構成的rdd。
takeordered
takeordered方法的輸入參數為一個整數n，它傳回一個由原rdd中前n小的元素構成的rdd。
top
top方法的輸入參數為一個整數n，它傳回一個由原rdd中前n大的元素構成的rdd。
fold
fold是一個高階方法，用于對原rdd的元素做彙總操作，彙總的時候使用一個自定義的初值和一個滿足結合律的二進制操作符。它首先在每一個rdd的分區中進行彙總，然後再彙總這些結果。
初值的取值取決于rdd中的元素類型和彙總操作的目的。比如，給定一個元素為整數的rdd，為了計算這個rdd中所有元素的和，初值取為0。相反，給定一個元素為整數的rdd，為了計算這個rdd中所有元素的乘積，初值則應取為1。
reduce
reduce是一個高階方法，用于對原rdd的元素做彙總操作，彙總的時候使用一個滿足結合律和交換律的二進制操作符。它類似于fold方法，然而，它并不需要初值。
鍵值對型rdd上的操作
鍵值對rdd上有一些額外的操作，我們在下面進行介紹。
countbykey
countbykey方法用于統計原rdd每個鍵的個數。它傳回一個map類執行個體，其中，鍵為原rdd中的鍵，值為個數。
lookup
lookup方法的輸入參數為一個鍵，傳回一個序列，這個序列的元素為原rdd中這個鍵對應的值。
數值型rdd上的操作
如果rdd的元素類型為integer、long、float或double，則這樣的rdd為數值型rdd。這類rdd還有一些對于統計分析十分有用的額外操作，下面将介紹一些常用的行動。
mean
mean方法傳回原rdd中元素的平均值。
stdev
stdev方法傳回原rdd中元素的标準差。
sum
sum方法傳回原rdd中所有元素的和。
variance
variance方法傳回原rdd中元素的方差。
3.5.5 儲存rdd
一般來說，資料處理完畢後，結果會儲存在硬碟上。spark允許開發者将rdd儲存在任何hadoop支援的存儲系統中。儲存在硬碟上的rdd可以被其他spark應用或hadoop應用使用。
本節介紹将rdd儲存成檔案的常用方法。
saveastextfile
saveastextfile方法将原rdd中的元素儲存在指定目錄中，這個目錄位于任何hadoop支援的存儲系統中。每一個rdd中的元素都用字元串表示并另存為文本中的一行。
saveasobjectfile
saveasobjectfile方法将原rdd中的元素序列化成java對象，存儲在指定目錄中。
saveassequencefile
saveassequencefile方法将鍵值對型rdd以sequencefile的格式儲存。鍵值對型rdd也可以以文本的格式儲存，隻須使用saveastextfile方法即可。
需要注意的是，上面的方法都把一個目錄的名字作為輸入參數，然後在這個目錄為每個rdd分區建立一個檔案。這種設計不僅高效而且可容錯。因為每一個分區被存成一個檔案，是以spark在儲存rdd的時候可以啟動多個任務，并行執行，将資料寫入檔案系統中。這樣也保證了寫入資料的過程是可容錯的。一旦有一個将分區寫入檔案的任務失敗了，spark可以再啟動一個任務，重寫剛才失敗任務建立的檔案。