結合了《Learning Spark: Lightning-Fast Big Data Analysis》和官方2.02版本的Spark Document總結了關於Spark概念性的一些知識。幫助大家對Spark有一個總體上的認知

一、Spark的兩個核心概念：

1、RDD：彈性分散式資料集
2、 Shared variables：共享變數

二、Spark元件：

Spark集成了很多元件。Spark的核心是一個計算引擎，用於誇多個worker machines或計算叢集排程，分佈和監視由很與計算任務組成的應用程式。

Spark緊密整合的優點：
1. 高層元件能夠受益於底層的改進。比如Spark核心優化後，它的SQL和ML庫也能提升速度。
2. 花費較少。包括部署、維修、測試、支援等的費用。一旦有新的元件加入到Spark棧中，所有使用Spark的機構都能夠立即使用這一新的元件。
3. Spark建立的應用能夠無縫組合其他的處理模型。比如，你用Spark寫一個利用ML演算法，不斷資料流中接收資料，對資料進行實時分類的程式。與此同時，分析師能夠通過SQL對同樣的資料進行實時的查詢操作。此外，資料工程師和資料科學家能夠通過Python Shell獲取同樣的書進行不同的分析。

Spark元件介紹：
1. Spark Core：包括Spark的基礎功能，比如任務排程元件、記憶體管理、故障恢復、儲存系統互動等。也是定義RDD（Spark最重要的程式設計概念）的API的所在地。
2. Spark SQL (structured data)：處理結構化資料的Spark包，利用SQL可以或者Hive進行資料查詢操作。它也支援很多資料來源，包括Hive表格，Parquet和JSON。Spark SQL除了給Spark提供了一個SQL介面外，還循序開發者將SQL查詢和RDD支援的可程式設計資料操作混合使用，從而進行更加複雜的分析。
3. Spark Streaming (real-time)：是一個能夠處理實時資料流的Spark元件。資料流包括生產Web伺服器產生的日誌檔案、web服務的使用者釋出的狀態更新的資訊。Spark Streaming提供了一個操作資料流的API，它與RDD API十分接近。API的底層設計能夠提供與Spark Core同等程度的容錯、吞吐量和擴充套件支援。
4. Mlib (machine learning)：提供了多種機器學習演算法，包括分類、迴歸、聚類、協同過濾、模型評價和資料匯入等等。也提供了ML的一些底層的處理，比如通用梯度下降優化演算法。這些演算法都能夠誇叢集擴充套件。
5. GraphX (processing)：用於處理影象（比如社交網路的朋友關係圖）和進行圖形平行計算的庫。與Spark Streaming和Spark SQL類似，GraphX也是Spark RDD API的擴充套件，允許我們建立一個有向圖，並給每個節點和邊任意設定屬性。GraphX也提供了豐富的影象處理運算元（例如subgraph和mapVertices）和常見的影象演算法庫（例如PageRank和triangle counting）。

Spark能夠在很多叢集管理器上執行，包括Hadoop YARN，Apache Mesos，和簡單的叢集管理器，包括Spark自己的Standalone Scheduler。

Core Spark概念
每個Spark應用都含有一個驅動程式，它將並行操作投放到叢集上。該驅動程式包括Spark應用的main主函式，在叢集上定義分散式資料集，並將其上實施各種操作。Spark shell本身就是一個驅動程式，因此我們能夠在shell中直接實施各種操作。如下圖所示。

三、初始化Spark

編寫Spark程式首先要做的一件事是建立一個SparkContext物件，這個物件是告訴Spark如何訪問一個叢集的。在建立SparkContext之前，需要先建立一個SparkConf物件，它包含了spark應用的資訊。
conf = SparkConf().setAppName(appName).setMaster(master)
sc = SparkContext(conf=conf)
其中，
appName： 應用的名字，會在叢集的UI上顯示
master：是一個Spark或者Mesos或者YARN叢集的URL，也可以是一個指示本地地址的string。
實際應用中，當在一個叢集上執行程式時，我們一般並不願意在程式中將master寫死，而是希望通過spark-submit將應用程式啟動的時候傳遞它。但在本地測試過程中可以使用local測試Spark。

3.1. 使用Shell
在Pyspark shell中，已經存在一個專有的直譯器可識別的SparkContext，叫做sc。如果再建立自己的SparkContext就無法工作。可以利用–matser命令設定Spark將要連線的叢集。可用利用–py-files語句將Python的.zip，.egg或.py檔案新增到runtime path，如果有多個檔案，用逗號隔開。也可以通過在–packages語句和maven在shell會話中新增依賴（例如SparkPackages）。任何附加的可能存在依賴的程式碼庫（例如SonaType）都可以用過–repositories語句新增。任何Spark Packages中存在的Python以來都必須通過pip命令手動安裝。下面舉幾個例子，在4個核上執行bin/pyspark：
./bin/pyspark–masterlocal[4]或者將code.py加入到搜索路徑（便於後邊使用importcode），利用下面的語句實現： /bin/pyspark –master local[4] –py-files code.py
如需檢視pyspark的所有操作，可以執行pyspark –help命令。事實上，pyspark呼叫了更加通用的spark -submit指令碼。

在IPython中也可以啟動PySpark會話，PySpark能夠在IPython 1.0.0以上版本上執行。如果要使用IPython，將變數PYSPARK_DRIVER_PYTHON設定為ipython就可以了。
PYSPARKDRIVERPYTHON=ipython./bin/pyspark類似的，如果要使用Jupyter，設置為： PYSPARK_DRIVER_PYTHON=jupyter ./bin/pyspark

四、彈性分散式資料集RDDs

兩種方法可以建立RDD：
（1）在現存的驅動程式中並行化集合
（2）引用一個外部儲存系統中的資料集，例如共享檔案系統，HDFS，HBase或者其他Hadoop輸入格式的資料來源

4.1. 並行化集合
並行化集合通過在已有的可迭代物件揮著集合上呼叫SparkContext中的parallelize實現。集合中的元素被拷貝後構成一個能夠進行並行操作的分散式資料集。例如建立一個包含數字1-5的並行化集合：
data = [1, 2, 3, 4, 5]
distData = sc.parallelize(data)
並行化的一個重要引數是並行的分割槽數目。Spark在叢集的每個分割槽上執行一個任務。一般來說，一個CPU一般會分2-4個分割槽。通常，Spark會根據叢集自動設定分割槽的數目。但也可以進行人為設定，通過設定sc.parallelize的第二個引數實現。注意，有些地方使用分片這個術語（等同於分片）來維護反向相容性。

4.2. 外部資料集
Spark可以從Hadoop支援的其他儲存單元建立分散式資料集，包括本地檔案系統，HDFS，Cassandra，HBase，Amazon S3等等。Spark支援文字檔案，檔案佇列和其他Hadoop輸入格式。
文字檔案RDD能夠通過SparkContext的textFile方法建立。該方法給每個檔案一個URI（一個計算機的本地地址，或是hdfs://，s3n://等URI），將檔案作為行的集合讀入。例如：

>>> distFile = sc.textFile(“data.txt”)

建立後，distFile可以通過資料集的操作進行很多活動。比如，計算檔案中的總行數可以通map和reduce操作實現：distFile.map(lambda s: len(s)).reduce(lambda a, b: a+b).
Spark讀取檔案的注意事項：
（1）如果使用本地檔案系統的路徑，那麼該檔案也必須能夠被其他工作節點獲取。或者將檔案拷貝到每個工作節點，或者使用安裝在網路上的共享檔案系統。
（2）textFile方法通過設定第二個引數能夠控制檔案的分割槽數目。預設情況下，Spark為檔案的每個block（HDFS預設一個block的大小為64MB）建立一個分割槽，但是可以通過設定引數獲得更多的分割槽數目。注意，分割槽數目不能少於block的數目。
除了文字檔案，Spark的Python API也支援其他一些資料格式：
（1）SparkContext.wholeTextFiles：能夠讀取包含多個小的文字檔案的字典，並返回每個（檔名filenames，內容content）對。這與textFile不同，textFile返回的是每個檔案中每行的一個記錄。
（2）RDD.saveAsPickleFile和SparkContext.pickleFile支援將RDD以一種簡單的格式儲存，這種格式由被pickle的python物件組成。批處理被用來進行pickle序列化，預設的大小為10。
（3）檔案佇列和Hadoop輸入/輸出格式。

注意：以上特性目前仍處於實驗狀態，主要針對高階使用者。未來，也可能被給予Spark SQL的read/write支援所替代，因為Spark SQL的方法更為人偏愛。

Writable支援
PySpark把Java鍵值對組成的RDD載入檔案佇列時，將writables轉化為Java格式，然後利用Pyrolite將Java物件進行pickle。當把鍵值對的一個RDD存入檔案佇列時，PySpark會進行一系列相反的操作。它會把Python物件unpickle成Java物件，並將它們轉化為writables。下列Writables能夠進行自動轉換：

比較意外的是，array型別並不支援。在讀寫時，使用者需要專門指定使用者的ArrayWritable子型別還需要。在寫入時，使用者還需要指定轉換器將array轉換成ArrayWritable子型別。在讀取時，預設的轉換器能夠將ArrayWritable子型別轉換成Java的Object[]，然後被pickle成配Python元組（tuples）。

儲存和載入檔案佇列
與文字檔案相似，檔案佇列能夠以特定的路徑進行儲存和載入操作。對於key和value類需要被特別說明，但對於標準的Writables並不需要。

>>> rdd = sc.parallelize(range(1, 4).map(lambda x: (x, "a"*x)))
>>> rdd.saveAsSequenceFile("path/to/file")
>>> sorted(sc.sequenceFile("path/to/file").collect())
[(1, 'a'), (2, 'aa'), (3, 'aaa')]

儲存和載入其他Hadoop輸入輸出格式
對於新舊版本的Hadoop Mapreduce API， Spark都支援任何Hadoop輸入格式的讀取和輸出格式的寫入。如果需要，Hadoop佈局能夠以Python字典的形式傳遞。