RDD論文翻譯：基於內存的集群計算容錯抽象

該論文來自Berkeley實驗室，英文標題為：Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing。下面的翻譯，我是基於科學網翻譯基礎上進行優化、修改、補充，這篇譯文翻譯得很不錯。在此基礎上，我增加了來自英文原文的圖和表格數據，以及譯文中缺少的未翻譯的部分。如果翻譯措辭或邏輯有誤，歡迎批評指正。

摘要

本文提出了分散式內存抽象的概念——彈性分散式數據集（RDD，Resilient Distributed Datasets），它具備像MapReduce等數據流模型的容錯特性，並且允許開發人員在大型集群上執行基於內存的計算。現有的數據流系統對兩種應用的處理並不高效：一是迭代式演算法，這在圖應用和機器學習領域很常見；二是互動式數據挖掘工具。這兩種情況下，將數據保存在內存中能夠極大地提高性能。為了有效地實現容錯，RDD提供了一種高度受限的共享內存，即RDD是只讀的，並且只能通過其他RDD上的批量操作來創建。儘管如此，RDD仍然足以表示很多類型的計算，包括MapReduce和專用的迭代編程模型（如Pregel）等。我們實現的RDD在迭代計算方面比Hadoop快20多倍，同時還可以在5-7秒內互動式地查詢1TB數據集。

1.引言

無論是工業界還是學術界，都已經廣泛使用高級集群編程模型來處理日益增長的數據，如MapReduce和Dryad。這些系統將分散式編程簡化為自動提供位置感知性調度、容錯以及負載均衡，使得大量用戶能夠在商用集群上分析超大數據集。

大多數現有的集群計算系統都是基於非循環的數據流模型。從穩定的物理存儲（如分散式文件系統）中載入記錄，記錄被傳入由一組確定性操作構成的DAG，然後寫回穩定存儲。DAG數據流圖能夠在運行時自動實現任務調度和故障恢復。

儘管非循環數據流是一種很強大的抽象方法，但仍然有些應用無法使用這種方式描述。我們就是針對這些不太適合非循環模型的應用，它們的特點是在多個並行操作之間重用工作數據集。這類應用包括：（1）機器學習和圖應用中常用的迭代演算法（每一步對數據執行相似的函數）；（2）互動式數據挖掘工具（用戶反覆查詢一個數據子集）。基於數據流的框架並不明確支持工作集，所以需要將數據輸出到磁碟，然後在每次查詢時重新載入，這帶來較大的開銷。

我們提出了一種分散式的內存抽象，稱為彈性分散式數據集（RDD，Resilient Distributed Datasets）。它支持基於工作集的應用，同時具有數據流模型的特點：自動容錯、位置感知調度和可伸縮性。RDD允許用戶在執行多個查詢時顯式地將工作集緩存在內存中，後續的查詢能夠重用工作集，這極大地提升了查詢速度。

RDD提供了一種高度受限的共享內存模型，即RDD是只讀的記錄分區的集合，只能通過在其他RDD執行確定的轉換操作（如map、join和group by）而創建，然而這些限制使得實現容錯的開銷很低。與分散式共享內存系統需要付出高昂代價的檢查點和回滾機制不同，RDD通過Lineage來重建丟失的分區：一個RDD中包含了如何從其他RDD衍生所必需的相關信息，從而不需要檢查點操作就可以重構丟失的數據分區。儘管RDD不是一個通用的共享內存抽象，但卻具備了良好的描述能力、可伸縮性和可靠性，但卻能夠廣泛適用於數據並行類應用。

第一個指出非循環數據流存在不足的並非是我們，例如，Google的Pregel[21]，是一種專門用於迭代式圖演算法的編程模型；Twister[13]和HaLoop[8]，是兩種典型的迭代式MapReduce模型。但是，對於一些特定類型的應用，這些系統提供了一個受限的通信模型。相比之下，RDD則為基於工作集的應用提供了更為通用的抽象，用戶可以對中間結果進行顯式的命名和物化，控制其分區，還能執行用戶選擇的特定操作（而不是在運行時去循環執行一系列MapReduce步驟）。RDD可以用來描述Pregel、迭代式MapReduce，以及這兩種模型無法描述的其他應用，如互動式數據挖掘工具（用戶將數據集裝入內存，然後執行ad-hoc查詢）。

Spark是我們實現的RDD系統，在我們內部能夠被用於開發多種並行應用。Spark採用Scala語言[5]實現，提供類似於DryadLINQ的集成語言編程介面[34]，使用戶可以非常容易地編寫並行任務。此外，隨著Scala新版本解釋器的完善，Spark還能夠用於互動式查詢大數據集。我們相信Spark會是第一個能夠使用有效、通用編程語言，並在集群上對大數據集進行互動式分析的系統。

我們通過微基準和用戶應用程序來評估RDD。實驗表明，在處理迭代式應用上Spark比Hadoop快高達20多倍，計算數據分析類報表的性能提高了40多倍，同時能夠在5-7秒的延時內互動式掃描1TB數據集。此外，我們還在Spark之上實現了Pregel和HaLoop編程模型（包括其位置優化策略），以庫的形式實現（分別使用了100和200行Scala代碼）。最後，利用RDD內在的確定性特性，我們還創建了一種Spark調試工具rddbg，允許用戶在任務期間利用Lineage重建RDD，然後像傳統調試器那樣重新執行任務。

本文首先在第2部分介紹了RDD的概念，然後第3部分描述Spark API，第4部分解釋如何使用RDD表示幾種並行應用（包括Pregel和HaLoop），第5部分討論Spark中RDD的表示方法以及任務調度器，第6部分描述具體實現和rddbg，第7部分對RDD進行評估，第8部分給出了相關研究工作，最後第9部分總結。

2.彈性分散式數據集（RDD）

本部分描述RDD和編程模型。首先討論設計目標（2.1），然後定義RDD（2.2），討論Spark的編程模型（2.3），並給出一個示例（2.4），最後對比RDD與分散式共享內存（2.5）。

2.1 目標和概述

我們的目標是為基於工作集的應用（即多個並行操作重用中間結果的這類應用）提供抽象，同時保持MapReduce及其相關模型的優勢特性：即自動容錯、位置感知性調度和可伸縮性。RDD比數據流模型更易於編程，同時基於工作集的計算也具有良好的描述能力。

在這些特性中，最難實現的是容錯性。一般來說，分散式數據集的容錯性有兩種方式：即數據檢查點和記錄數據的更新。我們面向的是大規模數據分析，數據檢查點操作成本很高：需要通過數據中心的網路連接在機器之間複製龐大的數據集，而網路帶寬往往比內存帶寬低得多，同時還需要消耗更多的存儲資源（在內存中複製數據可以減少需要緩存的數據量，而存儲到磁碟則會拖慢應用程序）。所以，我們選擇記錄更新的方式。但是，如果更新太多，那麼記錄更新成本也不低。因此，RDD只支持粗粒度轉換，即在大量記錄上執行的單個操作。將創建RDD的一系列轉換記錄下來（即Lineage），以便恢復丟失的分區。

雖然只支持粗粒度轉換限制了編程模型，但我們發現RDD仍然可以很好地適用於很多應用，特別是支持數據並行的批量分析應用，包括數據挖掘、機器學習、圖演算法等，因為這些程序通常都會在很多記錄上執行相同的操作。RDD不太適合那些非同步更新共享狀態的應用，例如並行web爬行器。因此，我們的目標是為大多數分析型應用提供有效的編程模型，而其他類型的應用交給專門的系統。

2.2 RDD抽象

RDD是只讀的、分區記錄的集合。RDD只能基於在穩定物理存儲中的數據集和其他已有的RDD上執行確定性操作來創建。這些確定性操作稱之為轉換，如map、filter、groupBy、join（轉換不是程開發人員在RDD上執行的操作）。

RDD不需要物化。RDD含有如何從其他RDD衍生（即計算）出本RDD的相關信息（即Lineage），據此可以從物理存儲的數據計算出相應的RDD分區。

2.3 編程模型

在Spark中，RDD被表示為對象，通過這些對象上的方法（或函數）調用轉換。

定義RDD之後，程序員就可以在動作中使用RDD了。動作是嚮應用程序返回值，或向存儲系統導出數據的那些操作，例如，count（返回RDD中的元素個數），collect（返回元素本身），save（將RDD輸出到存儲系統）。在Spark中，只有在動作第一次使用RDD時，才會計算RDD（即延遲計算）。這樣在構建RDD的時候，運行時通過管道的方式傳輸多個轉換。

程序員還可以從兩個方面控制RDD，即緩存和分區。用戶可以請求將RDD緩存，這樣運行時將已經計算好的RDD分區存儲起來，以加速後期的重用。緩存的RDD一般存儲在內存中，但如果內存不夠，可以寫到磁碟上。

另一方面，RDD還允許用戶根據關鍵字（key）指定分區順序，這是一個可選的功能。目前支持哈希分區和範圍分區。例如，應用程序請求將兩個RDD按照同樣的哈希分區方式進行分區（將同一機器上具有相同關鍵字的記錄放在一個分區），以加速它們之間的join操作。在Pregel和HaLoop中，多次迭代之間採用一致性的分區置換策略進行優化，我們同樣也允許用戶指定這種優化。

2.4 示例：控制台日誌挖掘

本部分我們通過一個具體示例來闡述RDD。假定有一個大型網站出錯，操作員想要檢查Hadoop文件系統（HDFS）中的日誌文件（TB級大小）來找出原因。通過使用Spark，操作員只需將日誌中的錯誤信息裝載到一組節點的內存中，然後執行互動式查詢。首先，需要在Spark解釋器中輸入如下Scala命令：

lines = spark.textFile("hdfs://...")

errors = lines.filter(_.startsWith("ERROR"))

errors.cache()

第1行從HDFS文件定義了一個RDD（即一個文本行集合），第2行獲得一個過濾後的RDD，第3行請求將errors緩存起來。注意在Scala語法中filter的參數是一個閉包。

這時集群還沒有開始執行任何任務。但是，用戶已經可以在這個RDD上執行對應的動作，例如統計錯誤消息的數目：

1errors.count()

用戶還可以在RDD上執行更多的轉換操作，並使用轉換結果，如：

使用errors的第一個action運行以後，Spark會把errors的分區緩存在內存中，極大地加快了後續計算速度。注意，最初的RDD lines不會被緩存。因為錯誤信息可能只佔原數據集的很小一部分（小到足以放入內存）。

最後，為了說明模型的容錯性，圖1給出了第3個查詢的Lineage圖。在lines RDD上執行filter操作，得到errors，然後再filter、map後得到新的RDD，在這個RDD上執行collect操作。Spark調度器以流水線的方式執行後兩個轉換，向擁有errors分區緩存的節點發送一組任務。此外，如果某個errors分區丟失，Spark只在相應的lines分區上執行filter操作來重建該errors分區。

圖1 示例中第三個查詢的Lineage圖。（方框表示RDD，箭頭表示轉換）

2.5 RDD與分散式共享內存

為了進一步理解RDD是一種分散式的內存抽象，表1列出了RDD與分散式共享內存（DSM，Distributed Shared Memory）[24]的對比。在DSM系統中，應用可以向全局地址空間的任意位置進行讀寫操作。（注意這裡的DSM，不僅指傳統的共享內存系統，還包括那些通過分散式哈希表或分散式文件系統進行數據共享的系統，比如Piccolo[28]）DSM是一種通用的抽象，但這種通用性同時也使得在商用集群上實現有效的容錯性更加困難。

RDD與DSM主要區別在於，不僅可以通過批量轉換創建（即「寫」）RDD，還可以對任意內存位置讀寫。也就是說，RDD限制應用執行批量寫操作，這樣有利於實現有效的容錯。特別地，RDD沒有檢查點開銷，因為可以使用Lineage來恢復RDD。而且，失效時只需要重新計算丟失的那些RDD分區，可以在不同節點上並行執行，而不需要回滾整個程序。

注意，通過備份任務的拷貝，RDD還可以處理落後任務（即運行很慢的節點），這點與MapReduce[12]類似。而DSM則難以實現備份任務，因為任務及其副本都需要讀寫同一個內存位置。

與DSM相比，RDD模型有兩個好處。第一，對於RDD中的批量操作，運行時將根據數據存放的位置來調度任務，從而提高性能。第二，對於基於掃描的操作，如果內存不足以緩存整個RDD，就進行部分緩存。把內存放不下的分區存儲到磁碟上，此時性能與現有的數據流系統差不多。

最後看一下讀操作的粒度。RDD上的很多動作（如count和collect）都是批量讀操作，即掃描整個數據集，可以將任務分配到距離數據最近的節點上。同時，RDD也支持細粒度操作，即在哈希或範圍分區的RDD上執行關鍵字查找。

3. Spark編程介面

Spark用Scala[5]語言實現了RDD的API。Scala是一種基於JVM的靜態類型、函數式、面向對象的語言。我們選擇Scala是因為它簡潔（特別適合互動式使用）、有效（因為是靜態類型）。但是，RDD抽象並不局限於函數式語言，也可以使用其他語言來實現RDD，比如像Hadoop[2]那樣用類表示用戶函數。

要使用Spark，開發者需要編寫一個driver程序，連接到集群以運行Worker，如圖2所示。Driver定義了一個或多個RDD，並調用RDD上的動作。Worker是長時間運行的進程，將RDD分區以Java對象的形式緩存在內存中。

圖2 Spark的運行時。用戶的driver程序啟動多個worker，worker從分散式文件系統中讀取數據塊，並將計算後的RDD分區緩存在內存中。

再看看2.4中的例子，用戶執行RDD操作時會提供參數，比如map傳遞一個閉包（closure，函數式編程中的概念）。Scala將閉包表示為Java對象，如果傳遞的參數是閉包，則這些對象被序列化，通過網路傳輸到其他節點上進行裝載。Scala將閉包內的變數保存為Java對象的欄位。例如，var x = 5; rdd.map(_ + x) 這段代碼將RDD中的每個元素加5。總的來說，Spark的語言集成類似於DryadLINQ。

RDD本身是靜態類型對象，由參數指定其元素類型。例如，RDD[int]是一個整型RDD。不過，我們舉的例子幾乎都省略了這個類型參數，因為Scala支持類型推斷。

雖然在概念上使用Scala實現RDD很簡單，但還是要處理一些Scala閉包對象的反射問題。如何通過Scala解釋器來使用Spark還需要更多工作，這點我們將在第6部分討論。不管怎樣，我們都不需要修改Scala編譯器。

3.1 Spark中的RDD操作

表2列出了Spark中的RDD轉換和動作。每個操作都給出了標識，其中方括弧表示類型參數。前面說過轉換是延遲操作，用於定義新的RDD；而動作啟動計算操作，並向用戶程序返回值或向外部存儲寫數據。

注意，有些操作只對鍵值對可用，比如join。另外，函數名與Scala及其他函數式語言中的API匹配，例如map是一對一的映射，而flatMap是將每個輸入映射為一個或多個輸出（與MapReduce中的map類似）。

除了這些操作以外，用戶還可以請求將RDD緩存起來。而且，用戶還可以通過Partitioner類獲取RDD的分區順序，然後將另一個RDD按照同樣的方式分區。有些操作會自動產生一個哈希或範圍分區的RDD，像groupByKey，reduceByKey和sort等。

4. 應用程序示例

現在我們講述如何使用RDD表示幾種基於數據並行的應用。首先討論一些迭代式機器學習應用（4.1），然後看看如何使用RDD描述幾種已有的集群編程模型，即MapReduce（4.2），Pregel（4.3），和Hadoop（4.4）。最後討論一下RDD不適合哪些應用（4.5）。

4.1 迭代式機器學習

很多機器學習演算法都具有迭代特性，運行迭代優化方法來優化某個目標函數，例如梯度下降方法。如果這些演算法的工作集能夠放入內存，將極大地加速程序運行。而且，這些演算法通常採用批量操作，例如映射和求和，這樣更容易使用RDD來表示。

例如下面的程序是邏輯回歸[15]的實現。邏輯回歸是一種常見的分類演算法，即尋找一個最佳分割兩組點（即垃圾郵件和非垃圾郵件）的超平面w。演算法採用梯度下降的方法：開始時w為隨機值，在每一次迭代的過程中，對w的函數求和，然後朝著優化的方向移動w。

首先定義一個名為points的緩存RDD，這是在文本文件上執行map轉換之後得到的，即將每個文本行解析為一個Point對象。然後在points上反覆執行map和reduce操作，每次迭代時通過對當前w的函數進行求和來計算梯度。7.1小節我們將看到這種在內存中緩存points的方式，比每次迭代都從磁碟文件裝載數據並進行解析要快得多。

已經在Spark中實現的迭代式機器學習演算法還有：kmeans（像邏輯回歸一樣每次迭代時執行一對map和reduce操作），期望最大化演算法（EM，兩個不同的map/reduce步驟交替執行），交替最小二乘矩陣分解和協同過濾演算法。Chu等人提出迭代式MapReduce也可以用來實現常用的學習演算法[11]。

4.2 使用RDD實現MapReduce

MapReduce模型[12]很容易使用RDD進行描述。假設有一個輸入數據集（其元素類型為T），和兩個函數myMap: T => List[(Ki, Vi)] 和 myReduce: (Ki; List[Vi]) ) List[R]，代碼如下：

data.flatMap(myMap).groupByKey().map((k, vs) => myReduce(k, vs))

如果任務包含combiner，則相應的代碼為：

data.flatMap(myMap).reduceByKey(myCombiner).map((k, v) => myReduce(k, v))

ReduceByKey操作在mapper節點上執行部分聚集，與MapReduce的combiner類似。

4.3 使用RDD實現Pregel

Pregel[21]是面向圖演算法的基於BSP範式[32]的編程模型。程序由一系列超步（Superstep）協調迭代運行。在每個超步中，各個頂點執行用戶函數，並更新相應的頂點狀態，變異圖拓撲，然後向下一個超步的頂點集發送消息。這種模型能夠描述很多圖演算法，包括最短路徑，雙邊匹配和PageRank等。

以PageRank為例介紹一下Pregel的實現。當前PageRank[7]記為r，頂點表示狀態。在每個超步中，各個頂點向其所有鄰居發送貢獻值r/n，這裡n是鄰居的數目。下一個超步開始時，每個頂點將其分值（rank）更新為 α/N + (1 - α) * Σci，這裡的求和是各個頂點收到的所有貢獻值的和，N是頂點的總數。

Pregel將輸入的圖劃分到各個worker上，並存儲在其內存中。在每個超步中，各個worker通過一種類似MapReduce的Shuffle操作交換消息。

Pregel的通信模式可以用RDD來描述，如圖3。主要思想是：將每個超步中的頂點狀態和要發送的消息存儲為RDD，然後根據頂點ID分組，進行Shuffle通信（即cogroup操作）。然後對每個頂點ID上的狀態和消息應用用戶函數（即mapValues操作），產生一個新的RDD，即(VertexID, (NewState, OutgoingMessages))。然後執行map操作分離出下一次迭代的頂點狀態和消息（即mapValues和flatMap操作）。代碼如下：

圖3 使用RDD實現Pregel時，一步迭代的數據流。（方框表示RDD，箭頭表示轉換）

需要注意的是，這種實現方法中，RDD grouped，newData和newVerts的分區方法與輸入RDD vertices一樣。所以，頂點狀態一直存在於它們開始執行的機器上，這跟原Pregel一樣，這樣就減少了通信成本。因為cogroup和mapValues保持了與輸入RDD相同的分區方法，所以分區是自動進行的。

完整的Pregel編程模型還包括其他工具，比如combiner，附錄A討論了它們的實現。下面將討論Pregel的容錯性，以及如何在實現相同容錯性的同時減少需要執行檢查點操作的數據量。

我們差不多用了100行Scala代碼在Spark上實現了一個類Pregel的API。7.2小節將使用PageRank演算法評估它的性能。

4.3.1 Pregel容錯

當前，Pregel基於檢查點機制來為頂點狀態及其消息實現容錯[21]。然而作者是這樣描述的：通過在其它的節點上記錄已發消息日誌，然後單獨重建丟失的分區，只需要恢復局部數據即可。上面提到這兩種方式，RDD都能夠很好地支持。

通過4.3小節的實現，Spark總是能夠基於Lineage實現頂點和消息RDD的重建，但是由於過長的Lineage鏈，恢復可能會付出高昂的代價。因為迭代RDD依賴於上一個RDD，對於部分分區來說，節點故障可能會導致這些分區狀態的所有迭代版本丟失，這就要求使用一種「級聯-重新執行」[20]的方式去依次重建每一個丟失的分區。為了避免這個問題，用戶可以周期性地在頂點和消息RDD上執行save操作，將狀態信息保存到持久存儲中。然後，Spark能夠在失敗的時候自動地重新計算這些丟失的分區（而不是回滾整個程序）。

最後，我們意識到，RDD也能夠實現檢查點數據的reduce操作，這要求通過一種高效的檢查點方案來表達檢查點數據。在很多Pregel作業中，頂點狀態都包括可變與不可變的組件，例如，在PageRank中，與一個頂點相鄰的頂點列表是不可變的，但是它們的排名是可變的，在這種情況下，我們可以使用一個來自可變數據的單獨RDD來替換不可變RDD，基於這樣一個較短的Lineage鏈，檢查點僅僅是可變狀態，圖4解釋了這種方式。

圖4 經過優化的Pregel使用RDD的數據流。可變狀態RDD必須設置檢查點，不可變狀態才可被快速重建。

在PageRank中，不可變狀態（相鄰頂點列表）遠大於可變狀態（浮點值），所以這種方式能夠極大地降低開銷。

4.4 使用RDD實現HaLoop

HaLoop[8]是Hadoop的一個擴展版本，它能夠改善具有迭代特性的MapReduce程序的性能。基於HaLoop編程模型的應用，使用reduce階段的輸出作為map階段下一輪迭代的輸入。它的循環感知任務調度器能夠保證，在每一輪迭代中處理同一個分區數據的連續map和reduce任務，一定能夠在同一台物理機上執行。確保迭代間locality特性，reduce數據在物理節點之間傳輸，並且允許數據緩存在本地磁碟而能夠被後續迭代重用。

使用RDD來優化HaLoop，我們在Spark上實現了一個類似HaLoop的API，這個庫只使用了200行Scala代碼。通過partitionBy能夠保證跨迭代的分區的一致性，每一個階段的輸入和輸出被緩存以用於後續迭代。

4.5 不適合使用RDD的應用

在2.1節我們討論過，RDD適用於具有批量轉換需求的應用，並且相同的操作作用於數據集的每一個元素上。在這種情況下，RDD能夠記住每個轉換操作，對應於Lineage圖中的一個步驟，恢復丟失分區數據時不需要寫日誌記錄大量數據。RDD不適合那些通過非同步細粒度地更新來共享狀態的應用，例如Web應用中的存儲系統，或者增量抓取和索引Web數據的系統，這樣的應用更適合使用一些傳統的方法，例如資料庫、RAMCloud[26]、Percolator[27]和Piccolo[28]。我們的目標是，面向批量分析應用的這類特定系統，提供一種高效的編程模型，而不是一些非同步應用程序。

5. RDD的描述及作業調度

我們希望在不修改調度器的前提下，支持RDD上的各種轉換操作，同時能夠從這些轉換獲取Lineage信息。為此，我們為RDD設計了一組小型通用的內部介面。

簡單地說，每個RDD都包含：（1）一組RDD分區（partition，即數據集的原子組成部分）；（2）對父RDD的一組依賴，這些依賴描述了RDD的Lineage；（3）一個函數，即在父RDD上執行何種計算；（4）元數據，描述分區模式和數據存放的位置。例如，一個表示HDFS文件的RDD包含：各個數據塊的一個分區，並知道各個數據塊放在哪些節點上。而且這個RDD上的map操作結果也具有同樣的分區，map函數是在父數據上執行的。表3總結了RDD的內部介面。

設計介面的一個關鍵問題就是，如何表示RDD之間的依賴。我們發現RDD之間的依賴關係可以分為兩類，即：（1）窄依賴（narrow dependencies）：子RDD的每個分區依賴於常數個父分區（即與數據規模無關）；（2）寬依賴（wide dependencies）：子RDD的每個分區依賴於所有父RDD分區。例如，map產生窄依賴，而join則是寬依賴（除非父RDD被哈希分區）。另一個例子見圖5。

圖5 窄依賴和寬依賴的例子。（方框表示RDD，實心矩形表示分區）

區分這兩種依賴很有用。首先，窄依賴允許在一個集群節點上以流水線的方式（pipeline）計算所有父分區。例如，逐個元素地執行map、然後filter操作；而寬依賴則需要首先計算好所有父分區數據，然後在節點之間進行Shuffle，這與MapReduce類似。第二，窄依賴能夠更有效地進行失效節點的恢復，即只需重新計算丟失RDD分區的父分區，而且不同節點之間可以並行計算；而對於一個寬依賴關係的Lineage圖，單個節點失效可能導致這個RDD的所有祖先丟失部分分區，因而需要整體重新計算。

通過RDD介面，Spark只需要不超過20行代碼實現便可以實現大多數轉換。5.1小節給出了例子，然後我們討論了怎樣使用RDD介面進行調度（5.2），最後討論一下基於RDD的程序何時需要數據檢查點操作（5.3）。

5.1 RDD實現舉例

HDFS文件：目前為止我們給的例子中輸入RDD都是HDFS文件，對這些RDD可以執行：partitions操作返回各個數據塊的一個分區（每個Partition對象中保存數據塊的偏移），preferredLocations操作返回數據塊所在的節點列表，iterator操作對數據塊進行讀取。

map：任何RDD上都可以執行map操作，返回一個MappedRDD對象。該操作傳遞一個函數參數給map，對父RDD上的記錄按照iterator的方式執行這個函數，並返回一組符合條件的父RDD分區及其位置。

union：在兩個RDD上執行union操作，返回兩個父RDD分區的並集。通過相應父RDD上的窄依賴關係計算每個子RDD分區（注意union操作不會過濾重複值，相當於SQL中的UNION ALL）。

sample：抽樣與映射類似，但是sample操作中，RDD需要存儲一個隨機數產生器的種子，這樣每個分區能夠確定哪些父RDD記錄被抽樣。

join：對兩個RDD執行join操作可能產生窄依賴（如果這兩個RDD擁有相同的哈希分區或範圍分區），可能是寬依賴，也可能兩種依賴都有（比如一個父RDD有分區，而另一父RDD沒有）。

5.2 Spark任務調度器

調度器根據RDD的結構信息為每個動作確定有效的執行計劃。調度器的介面是runJob函數，參數為RDD及其分區集，和一個RDD分區上的函數。該介面足以表示Spark中的所有動作（即count、collect、save等）。

總的來說，我們的調度器跟Dryad類似，但我們還考慮了哪些RDD分區是緩存在內存中的。調度器根據目標RDD的Lineage圖創建一個由stage構成的無迴路有向圖（DAG）。每個stage內部儘可能多地包含一組具有窄依賴關係的轉換，並將它們流水線並行化（pipeline）。stage的邊界有兩種情況：一是寬依賴上的Shuffle操作；二是已緩存分區，它可以縮短父RDD的計算過程。例如圖6。父RDD完成計算後，可以在stage內啟動一組任務計算丟失的分區。

圖6 Spark怎樣劃分任務階段（stage）的例子。實線方框表示RDD，實心矩形表示分區（黑色表示該分區被緩存）。要在RDD G上執行一個動作，調度器根據寬依賴創建一組stage，並在每個stage內部將具有窄依賴的轉換流水線化（pipeline）。 本例不用再執行stage 1，因為B已經存在於緩存中了，所以只需要運行2和3。

調度器根據數據存放的位置分配任務，以最小化通信開銷。如果某個任務需要處理一個已緩存分區，則直接將任務分配給擁有這個分區的節點。否則，如果需要處理的分區位於多個可能的位置（例如，由HDFS的數據存放位置決定），則將任務分配給這一組節點。

對於寬依賴（例如需要Shuffle的依賴），目前的實現方式是，在擁有父分區的節點上將中間結果物化，簡化容錯處理，這跟MapReduce中物化map輸出很像。

如果某個任務失效，只要stage中的父RDD分區可用，則只需在另一個節點上重新運行這個任務即可。如果某些stage不可用（例如，Shuffle時某個map輸出丟失），則需要重新提交這個stage中的所有任務來計算丟失的分區。

最後，lookup動作允許用戶從一個哈希或範圍分區的RDD上，根據關鍵字讀取一個數據元素。這裡有一個設計問題。Driver程序調用lookup時，只需要使用當前調度器介面計算關鍵字所在的那個分區。當然任務也可以在集群上調用lookup，這時可以將RDD視為一個大的分散式哈希表。這種情況下，任務和被查詢的RDD之間的並沒有明確的依賴關係（因為worker執行的是lookup），如果所有節點上都沒有相應的緩存分區，那麼任務需要告訴調度器計算哪些RDD來完成查找操作。

5.3 檢查點

儘管RDD中的Lineage信息可以用來故障恢復，但對於那些Lineage鏈較長的RDD來說，這種恢復可能很耗時。例如4.3小節中的Pregel任務，每次迭代的頂點狀態和消息都跟前一次迭代有關，所以Lineage鏈很長。如果將Lineage鏈存到物理存儲中，再定期對RDD執行檢查點操作就很有效。

一般來說，Lineage鏈較長、寬依賴的RDD需要採用檢查點機制。這種情況下，集群的節點故障可能導致每個父RDD的數據塊丟失，因此需要全部重新計算[20]。將窄依賴的RDD數據存到物理存儲中可以實現優化，例如前面4.1小節邏輯回歸的例子，將數據點和不變的頂點狀態存儲起來，就不再需要檢查點操作。

當前Spark版本提供檢查點API，但由用戶決定是否需要執行檢查點操作。今後我們將實現自動檢查點，根據成本效益分析確定RDD Lineage圖中的最佳檢查點位置。

值得注意的是，因為RDD是只讀的，所以不需要任何一致性維護（例如寫複製策略，分散式快照或者程序暫停等）帶來的開銷，後台執行檢查點操作。

我們使用10000行Scala代碼實現了Spark。系統可以使用任何Hadoop數據源（如HDFS，Hbase）作為輸入，這樣很容易與Hadoop環境集成。Spark以庫的形式實現，不需要修改Scala編譯器。

這裡討論關於實現的三方面問題：（1）修改Scala解釋器，允許交互模式使用Spark（6.1）；（2）緩存管理（6.2）；（3）調試工具rddbg（6.3）。

6. 實現

6.1 解釋器的集成

像Ruby和Python一樣，Scala也有一個互動式shell。基於內存的數據可以實現低延時，我們希望允許用戶從解釋器互動式地運行Spark，從而在大數據集上實現大規模並行數據挖掘。

Scala解釋器通常根據將用戶輸入的代碼行，來對類進行編譯，接著裝載到JVM中，然後調用類的函數。這個類是一個包含輸入行變數或函數的單例對象，並在一個初始化函數中運行這行代碼。例如，如果用戶輸入代碼var x = 5，接著又輸入println(x)，則解釋器會定義一個包含x的Line1類，並將第2行編譯為println(Line1.getInstance().x)。

在Spark中我們對解釋器做了兩點改動：

1. 類傳輸：解釋器能夠支持基於HTTP傳輸類位元組碼，這樣worker節點就能獲取輸入每行代碼對應的類的位元組碼。
2. 改進的代碼生成邏輯：通常每行上創建的單態對象通過對應類上的靜態方法進行訪問。也就是說，如果要序列化一個閉包，它引用了前面代碼行中變數，比如上面的例子Line1.x，Java不會根據對象關係傳輸包含x的Line1實例。所以worker節點不會收到x。我們將這種代碼生成邏輯改為直接引用各個行對象的實例。圖7說明了解釋器如何將用戶輸入的一組代碼行解釋為Java對象。

圖7 Spark解釋器如何將用戶輸入的兩行代碼解釋為Java對象

Spark解釋器便於跟蹤處理大量對象關係引用，並且便利了HDFS數據集的研究。我們計劃以Spark解釋器為基礎，開發提供高級數據分析語言支持的互動式工具，比如類似SQL和Matlab。

6.2 緩存管理

Worker節點將RDD分區以Java對象的形式緩存在內存中。由於大部分操作是基於掃描的，採取RDD級的LRU（最近最少使用）替換策略（即不會為了裝載一個RDD分區而將同一RDD的其他分區替換出去）。目前這種簡單的策略適合大多數用戶應用。另外，使用帶參數的cache操作可以設定RDD的緩存優先順序。

6.3 rddbg：RDD程序的調試工具

RDD的初衷是為了實現容錯以能夠再計算（re-computation），這個特性使得調試更容易。我們創建了一個名為rddbg的調試工具，它是通過基於程序記錄的Lineage信息來實現的，允許用戶：（1）重建任何由程序創建的RDD，並執行互動式查詢；（2）使用一個單進程Java調試器（如jdb）傳入計算好的RDD分區，能夠重新運行作業中的任何任務。

我們強調一下，rddbg不是一個完全重放的調試器：特別是不對非確定性的代碼或動作進行重放。但如果某個任務一直運行很慢（比如由於數據分布不均勻或者異常輸入等原因），仍然可以用它來幫助找到其中的邏輯錯誤和性能問題。

舉個例子，我們使用rddbg去解決用戶Spam分類作業中的一個bug，這個作業中的每次迭代都產生0值。在調試器中重新執行reduce任務，很快就能發現，輸入的權重向量（存儲在一個用戶自定義的向量類中）竟然是空值。由於從一個未初始化的稀疏向量中讀取總是返回0，運行時也不會拋出異常。在這個向量類中設置一個斷點，然後運行這個任務，引導程序很快就運行到設置的斷點處，我們發現向量類的一個數組欄位的值為空，我們診斷出了這個bug：稀疏向量類中的數據欄位被錯誤地使用transient來修飾，導致序列化時忽略了該欄位的數據。

rddbg給程序執行帶來的開銷很小。程序本來就需要將各個RDD中的所有閉包序列化並通過網路傳送，只不過使用rddbg同時還要將這些閉集記錄到磁碟。

7. 評估

我們在Amazon EC2[1]上進行了一系列實驗來評估Spark及RDD的性能，並與Hadoop及其他應用程序的基準進行了對比。總的說來，結果如下：

（1）對於迭代式機器學習應用，Spark比Hadoop快20多倍。這種加速比是因為：數據存儲在內存中，同時Java對象緩存避免了反序列化操作。

（2）用戶編寫的應用程序執行結果很好。例如，Spark分析報表比Hadoop快40多倍。

（3）如果節點發生失效，通過重建那些丟失的RDD分區，Spark能夠實現快速恢復。

（4）Spark能夠在5-7s延時範圍內，互動式地查詢1TB大小的數據集。

我們基準測試首先從一個運行在Hadoop上的具有迭代特徵的機器學習應用（7.1）和PageRank（7.2）開始，然後評估在Spark中當工作集不能適應緩存（7.4）時系統容錯恢復能力（7.3），最後討論用戶應用程序（7.5）和互動式數據挖掘（7.6）的結果。

除非特殊說明，我們的實驗使用m1.xlarge EC2 節點，4核15GB內存，使用HDFS作為持久存儲，塊大小為256M。在每個作業運行執行時，為了保證磁碟讀時間更加精確，我們清理了集群中每個節點的操作系統緩存。

7.1 可迭代的機器學習應用

我們實現了2個迭代式機器學習（ML）應用，Logistic回歸和K-means演算法，與如下系統進行性能對比：

• Hadoop：Hadoop 0.20.0穩定版。
• HadoopBinMem：在首輪迭代中執行預處理，通過將輸入數據轉換成為開銷較低的二進位格式來減少後續迭代過程中文本解析的開銷，在HDFS中載入到內存。
• Spark：基於RDD的系統，在首輪迭代中緩存Java對象以減少後續迭代過程中解析、反序列化的開銷。

我們使用同一數據集在相同條件下運行Logistic回歸和K-means演算法：使用400個任務（每個任務處理的輸入數據塊大小為256M），在25-100台機器，執行10次迭代處理100G輸入數據集（表4）。兩個作業的關鍵區別在於每輪迭代單個位元組的計算量不同。K-means的迭代時間取決於更新聚類坐標耗時，Logistic回歸是非計算密集型的，但是在序列化和解析過程中非常耗時。

由於典型的機器學習演算法需要數10輪迭代，然後再合併，我們分別統計了首輪迭代和後續迭代計算的耗時，並從中發現，在內存中緩存RDD極大地加快了後續迭代的速度。

首輪迭代。在首輪迭代過程中，三個系統都是從HDFS中讀取文本數據作為輸入。圖9中「First Iteration」顯示了首輪迭代的柱狀圖，實驗中Spark快於Hadoop，主要是因為Hadoop中的各個分散式組件基於心跳協議來發送信號帶來了開銷。HadoopBinMem是最慢的，因為它通過一個額外的MapReduce作業將數據轉換成二進位格式。

圖8 首輪迭代後Hadoop、HadoopBinMen、Spark運行時間對比

後續迭代。圖9顯示了後續迭代的平均耗時，圖8對比了不同聚類大小條件下耗時情況，我們發現在100個節點上運行Logistic回歸程序，Spark比Hadoop、HadoopBinMem分別快25.3、20.7倍。從圖8（b）可以看到，Spark僅僅比Hadoop、HadoopBinMem分別快1.9、3.2倍，這是因為K-means程序的開銷取決於計算（用更多的節點有助於提高計算速度的倍數）。

後續迭代中，Hadoop仍然從HDFS讀取文本數據作為輸入，所以從首輪迭代開始Hadoop的迭代時間並沒有明顯的改善。使用預先轉換的SequenceFile文件（Hadoop內建的二進位文件格式），HadoopBinMem在後續迭代中節省了解析的代價，但是仍然帶來的其他的開銷，如從HDFS讀SequenceFile文件並轉換成Java對象。因為Spark直接讀取緩存於RDD中的Java對象，隨著聚類尺寸的線性增長，迭代時間大幅下降。

圖9：首輪及其後續迭代平均時間對比

理解速度提升。我們非常驚奇地發現，Spark甚至勝過了基於內存存儲二進位數據的Hadoop（HadoopBinMem），幅度高達20倍之多，Hadoop運行慢是由於如下幾個原因：

1. Hadoop軟體棧的最小開銷
2. 讀數據時HDFS棧的開銷
3. 將二進位記錄轉換成內存Java對象的代價

為了估測1，我們運行空的Hadoop作業，僅僅執行作業的初始化、啟動任務、清理工作就至少耗時25秒。對於2，我們發現為了服務每一個HDFS數據塊，HDFS進行了多次複製以及計算校驗和操作。

為了估測3，我們在單個節點上運行了微基準程序，在輸入的256M數據上計算Logistic回歸，結果如表5所示。首先，在內存中的HDFS文件和本地文件的不同導致通過HDFS介面讀取耗時2秒，甚至數據就在本地內存中。其次，文本和二進位格式輸入的不同造成了解析耗時7秒的開銷。最後，預解析的二進位文件轉換為內存中的Java對象，耗時3秒。每個節點處理多個塊時這些開銷都會累積起來，然而通過緩存RDD作為內存中的Java對象，Spark只需要耗時3秒。

表5 Logistic回歸迭代時間內存中的HDFS文件內存中的本地文件緩存的RDD文本輸入

7.2 PageRank

通過使用存儲在HDFS上的49G Wikipedia導出數據，我們比較了使用RDD實現的Pregel與使用Hadoop計算PageRank的性能。PageRank演算法通過10輪迭代處理了大約400萬文章的鏈接圖數據，圖10顯示了在30個節點上，Spark處理速度是Hadoop的2倍多，改進後對輸入進行Hash分區速度提升到2.6倍，使用Combiner後提升到3.6倍，這些結果數據也隨著節點擴展到60個時同步放大。

圖10 迭代時間對比

7.3 容錯恢復

基於K-means演算法應用程序，我們評估了在單點故障（SPOF）時使用Lneage信息創建RDD分區的開銷。圖11顯示了，K-means應用程序運行在75個節點的集群中進行了10輪迭代，我們在正常操作和進行第6輪迭代開始時一個節點發生故障的情況下對耗時進行了對比。沒有任何失敗，每輪迭代啟動了400個任務處理100G數據。

圖11 SPOF時K-means應用程序迭代時間

第5輪迭代結束時大約耗時58秒，第6輪迭代時Kill掉一個節點，該節點上的任務都被終止（包括緩存的分區數據）。Spark調度器調度這些任務在其他節點上重新並行運行，並且重新讀取基於Lineage信息重建的RDD輸入數據並進行緩存，這使得迭代計算耗時增加到80秒。一旦丟失的RDD分區被重建，平均迭代時間又回落到58秒。

7.4 內存不足時表現

到現在為止，我們能保證集群中的每個節點都有足夠的內存去緩存迭代過程中使用的RDD，如果沒有足夠的內存來緩存一個作業的工作集，Spark又是如何運行的呢？在實驗中，我們通過在每個節點上限制緩存RDD所需要的內存資源來配置Spark，在不同的緩存配置條件下執行Logistic回歸，結果如圖12。我們可以看出，隨著緩存的減小，性能平緩地下降。

圖12 Spark上運行Logistic回歸的性能表現

7.5 基於Spark構建的用戶應用程序

In-Memory分析。視頻分發公司Conviva使用Spark極大地提升了為客戶處理分析報告的速度，以前基於Hadoop使用大約20個Hive[3]查詢來完成，這些查詢作用在相同的數據子集上（滿足用戶提供的條件），但是在不同分組的欄位上執行聚合操作（SUM、AVG、COUNT DISTINCT等）需要使用單獨的MapReduce作業。該公司使用Spark只需要將相關數據載入到內存中一次，然後運行上述聚合操作，在Hadoop集群上處理200G壓縮數據並生成報耗時20小時，而使用Spark基於96G內存的2個節點耗時30分鐘即可完成，速度提升40倍，主要是因為不需要再對每個作業重複地執行解壓縮和過濾操作。

城市交通建模。在Berkeley的Mobile Millennium項目[17]中，基於一系列分散的汽車GPS監測數據，研究人員使用並行化機器學習演算法來推算公路交通擁堵狀況。數據來自市區10000個互聯的公路線路網，還有600000個由汽車GPS裝置採集到的樣本數據，這些數據記錄了汽車在兩個地點之間行駛的時間（每一條路線的行駛時間可能跨多個公路線路網）。使用一個交通模型，通過推算跨多個公路網行駛耗時預期，系統能夠估算擁堵狀況。研究人員使用Spark實現了一個可迭代的EM演算法，其中包括向Worker節點廣播路線網路信息，在E和M階段之間執行reduceByKey操作，應用從20個節點擴展到80個節點（每個節點4核），如圖13（a）所示：

圖13 每輪迭代運行時間（a）交通建模應用程序（b）基於Spark的社交網路的Spam分類 社交網路Spam分類。Berkeley的Monarch項目[31]使用Spark識別Twitter消息上的Spam鏈接。他們在Spark上實現了一個類似7.1小節中示例的Logistic回歸分類器，不同的是使用分散式的reduceByKey操作並行對梯度向量求和。圖13（b）顯示了基於50G數據子集訓練訓練分類器的結果，整個數據集是250000的URL、至少10^7個與網路相關的特徵/維度，內容、詞性與訪問一個URL的頁面相關。隨著節點的增加，這並不像交通應用程序那樣近似線性，主要是因為每輪迭代的固定通信代價較高。

7.6 互動式數據挖掘

為了展示Spark互動式處理大數據集的能力，我們在100個m2.4xlarge EC2實例（8核68G內存）上使用Spark分析1TB從2008-10到2009-4這段時間的Wikipedia頁面瀏覽日誌數據，在整個輸入數據集上簡單地查詢如下內容以獲取頁面瀏覽總數：（1）全部頁面；（2）頁面的標題能精確匹配給定的關鍵詞；（3）頁面的標題能部分匹配給定的關鍵詞。

圖14 顯示了分別在整個、1/2、1/10的數據上查詢的響應時間，甚至1TB數據在Spark上查詢僅耗時5-7秒，這比直接操作磁碟數據快幾個數量級。例如，從磁碟上查詢1TB數據耗時170秒，這表明了RDD緩存使得Spark成為一個互動式數據挖掘的強大工具。

8. 相關工作

分散式共享內存（DSM）。RDD可以看成是一個基於DSM研究[24]得到的抽象。在2.5節我們討論過，RDD提供了一個比DSM限制更嚴格的編程模型，並能在節點失效時高效地重建數據集。DSM通過檢查點[19]實現容錯，而Spark使用Lineage重建RDD分區，這些分區可以在不同的節點上重新並行處理，而不需要將整個程序回退到檢查點再重新運行。RDD能夠像MapReduce一樣將計算推向數據[12]，並通過推測執行來解決某些任務計算進度落後的問題，推測執行在一般的DSM系統上是很難實現的。

In-Memory集群計算。Piccolo[28]是一個基於可變的、In-Memory的分散式表的集群編程模型。因為Piccolo允許讀寫表中的記錄，它具有與DSM類似的恢復機制，需要檢查點和回滾，但是不能推測執行，也沒有提供類似groupBy、sort等更高級別的數據流運算元，用戶只能直接讀取表單元數據來實現。可見，Piccolo是比Spark更低級別的編程模型，但是比DSM要高級。

RAMClouds[26]適合作為Web應用的存儲系統，它同樣提供了細粒度讀寫操作，所以需要通過記錄日誌來實現容錯。

數據流系統。RDD借鑒了DryadLINQ[34]、Pig[25]和FlumeJava[9]的「並行收集」編程模型，通過允許用戶顯式地將未序列化的對象保存在內存中，以此來控制分區和基於key隨機查找，從而有效地支持基於工作集的應用。RDD保留了那些數據流系統更高級別的編程特性，這對那些開發人員來說也比較熟悉，而且，RDD也能夠支持更多類型的應用。RDD新增的擴展，從概念上看很簡單，其中Spark是第一個使用了這些特性的系統，類似DryadLINQ編程模型，能夠有效地支持基於工作集的應用。

面向基於工作集的應用，已經開發了一些專用系統，像Twister[13]、HaLoop[8]實現了一個支持迭代的MapReduce模型；Pregel[21]，支持圖應用的BSP計算模型。RDD是一個更通用的抽象，它能夠描述支持迭代的MapReduce、Pregel，還有現有一些系統未能處理的應用，如互動式數據挖掘。特別地，它能夠讓開發人員動態地選擇操作來運行在RDD上（如查看查詢的結果以決定下一步運行哪個查詢），而不是提供一系列固定的步驟去執行迭代，RDD還支持更多類型的轉換。

最後，Dremel[22]是一個低延遲查詢引擎，它面向基於磁碟存儲的大數據集，這類數據集是把嵌套記錄數據生成基於列的格式。這種格式的數據也能夠保存為RDD並在Spark系統中使用，但Spark也具備將數據載入到內存來實現快速查詢的能力。

Lineage。我們通過參考[6]到[10]做過調研，在科學計算和資料庫領域，對於一些應用，如需要解釋結果以及允許被重新生成、工作流中發現了bug或者數據集丟失需要重新處理數據，表示數據的Lineage和原始信息一直以來都是一個研究課題。RDD提供了一個受限的編程模型，在這個模型中使用細粒度的Lineage來表示是非常容易的，因此它可以被用於容錯。

緩存系統。Nectar[14]能夠通過識別帶有程序分析的子表達式，跨DryadLINQ作業重用中間結果，如果將這種能力加入到基於RDD的系統會非常有趣。但是Nectar並沒有提供In-Memory緩存，也不能夠讓用戶顯式地控制應該緩存那個數據集，以及如何對其進行分區。Ciel[23]同樣能夠記住任務結果，但不能提供In-Memory緩存並顯式控制它。

語言迭代。DryadLINQ[34]能夠使用LINQ獲取到表達式樹然後在集群上運行，Spark系統的語言集成與它很類似。不像DryadLINQ，Spark允許用戶顯式地跨查詢將RDD存儲到內存中，並通過控制分區來優化通信。Spark支持互動式處理，但DryadLINQ卻不支持。

關係資料庫。從概念上看，RDD類似於資料庫中的視圖，緩存RDD類似於物化視圖[29]。然而，資料庫像DSM系統一樣，允許典型地讀寫所有記錄，通過記錄操作和數據的日誌來實現容錯，還需要花費額外的開銷來維護一致性。RDD編程模型通過增加更多限制來避免這些開銷。

9. 總結

我們提出的RDD是一個面向，運行在普通商用機集群之上並行數據處理應用的分散式內存抽象。RDD廣泛支持基於工作集的應用，包括迭代式機器學習和圖演算法，還有互動式數據挖掘，然而它保留了數據流模型中引人注目的特點，如自動容錯恢復，處理執行進度落後的任務，以及感知調度。它是通過限制編程模型，進而允許高效地重建RDD分區來實現的。RDD實現處理迭代式作業的速度超過Hadoop大約20倍，而且還能夠互動式查詢數百G數據。

致謝

首先感謝Spark用戶，包括Timothy Hunter、Lester Mackey、Dilip Joseph、Jibin Zhan和Teodor Moldovan，他們在真實的應用中使用Spark，提出了寶貴的建議，同時也發現了一些新的研究挑戰。這次研究離不開以下組織或團體的大力支持：Berkeley AMP Lab創立贊助者Google和SAP，AMP Lab贊助者Amazon Web Services、Cloudera、Huawei、IBM、Intel、Microsoft、NEC、NetApp和VMWare，國家配套資金加州MICRO項目（助學金 06-152，07-010），國家自然科學基金 （批准 CNS-0509559），加州大學工業/大學合作研究項目 （UC Discovery）授予的COM07-10240，以及自然科學和加拿大工程研究理事會。

參考

[1] Amazon EC2. http://aws.amazon.com/ec2.

[2] Apache Hadoop. http://hadoop.apache.org.

[3] Apache Hive. http://hadoop.apache.org/hive.

[4] Applications powered by Hadoop. http://wiki.apache.org/hadoop/PoweredBy.

[5] Scala. http://www.scala-lang.org.

[6] R. Bose and J. Frew. Lineage retrieval for scientific data processing: a survey. ACM Computing Surveys, 37:1–28,

2005.

[7] S. Brin and L. Page. The anatomy of a large-scale hypertextual web search engine. In WWW, 1998.

[8] Y. Bu, B. Howe, M. Balazinska, and M. D. Ernst. HaLoop: efficient iterative data processing on large clusters. Proc. VLDB Endow., 3:285–296, September 2010.

[9] C. Chambers, A. Raniwala, F. Perry, S. Adams, R. R. Henry, R. Bradshaw, and N. Weizenbaum. Flumejava: easy, efficient data-parallel pipelines. In Proceedings of the 2010 ACM SIGPLAN conference on Programming language design and implementation, PLDI 』10. ACM, 2010.

[10] J. Cheney, L. Chiticariu, and W.-C. Tan. Provenance in databases: Why, how, and where. Foundations and Trends in Databases, 1(4):379–474, 2009.

[11] C. T. Chu, S. K. Kim, Y. A. Lin, Y. Yu, G. R. Bradski, A. Y. Ng, and K. Olukotun. Map-reduce for machine learning on multicore. In NIPS 』06, pages 281–288. MIT Press, 2006.

[12] J. Dean and S. Ghemawat. MapReduce: Simplified data processing on large clusters. In OSDI, 2004.

[13] J. Ekanayake, H. Li, B. Zhang, T. Gunarathne, S.-H. Bae, J. Qiu, and G. Fox. Twister: a runtime for iterative mapreduce. In HPDC 』10, 2010.

[14] P. K. Gunda, L. Ravindranath, C. A. Thekkath, Y. Yu, and L. Zhuang. Nectar: automatic management of data and computation in datacenters. In OSDI 』10, 2010.

[15] T. Hastie, R. Tibshirani, and J. Friedman. The Elements of Statistical Learning: Data Mining, Inference, and Prediction. Springer Publishing Company, New York, NY, 2009.

[16] U. Hoelzle and L. A. Barroso. The Datacenter as a Computer: An Introduction to the Design of Warehouse-Scale Machines. Morgan and Claypool Publishers, 1st edition, 2009.

[17] Mobile Millennium Project. http://traffic.berkeley.edu.

[18] M. Isard, M. Budiu, Y. Yu, A. Birrell, and D. Fetterly. Dryad: distributed data-parallel programs from sequential building blocks. In EuroSys 07, 2007.

[19] A.-M. Kermarrec, G. Cabillic, A. Gefflaut, C. Morin, and I. Puaut. A recoverable distributed shared memory integrating coherence and recoverability. In FTCS 』95, 1995.

[20] S. Y. Ko, I. Hoque, B. Cho, and I. Gupta. On availability of intermediate data in cloud computations. In HotOS

』09, 2009.

[21] G. Malewicz, M. H. Austern, A. J. Bik, J. C. Dehnert, I. Horn, N. Leiser, and G. Czajkowski. Pregel: a system for large-scale graph processing. In SIGMOD, pages 135–146, 2010.

[22] S. Melnik, A. Gubarev, J. J. Long, G. Romer, S. Shivakumar, M. Tolton, and T. Vassilakis. Dremel: interactive analysis of web-scale datasets. Proc. VLDB Endow., 3:330–339, Sept 2010.

[23] D. G. Murray, M. Schwarzkopf, C. Smowton, S. Smith, A. Madhavapeddy, and S. Hand. Ciel: a universal execution engine for distributed data-flow computing. In NSDI, 2011.

[24] B. Nitzberg and V. Lo. Distributed shared memory: a survey of issues and algorithms. Computer, 24(8):52–60, aug 1991.

[25] C. Olston, B. Reed, U. Srivastava, R. Kumar, and A. Tomkins. Pig latin: a not-so-foreign language for data processing. In SIGMOD 』08, pages 1099–1110.

[26] J. Ousterhout, P. Agrawal, D. Erickson, C. Kozyrakis, J. Leverich, D. Mazi ` eres, S. Mitra, A. Narayanan, G. Parulkar, M. Rosenblum, S. M. Rumble, E. Stratmann, and R. Stutsman. The case for RAMClouds: scalable high-performance storage entirely in dram. SIGOPS Oper. Syst. Rev., 43:92–105, Jan 2010.

[27] D. Peng and F. Dabek. Large-scale incremental processing using distributed transactions and notifications. In OSDI 2010.

[28] R. Power and J. Li. Piccolo: Building fast, distributed programs with partitioned tables. In Proc. OSDI 2010,

2010.

[29] R. Ramakrishnan and J. Gehrke. Database Management Systems. McGraw-Hill, Inc., 3 edition, 2003.

[30] D. Spiewak and T. Zhao. ScalaQL: Language-integrated database queries for scala. In SLE, pages 154–163, 2009.

[31] K. Thomas, C. Grier, J. Ma, V. Paxson, and D. Song. Design and evaluation of a real-time URL spam filtering service. In IEEE Symposium on Security and Privacy, 2011.

[32] L. G. Valiant. A bridging model for parallel computation. Commun. ACM, 33:103–111, August 1990.

[33] J. W. Young. A first order approximation to the optimum checkpoint interval. Commun. ACM, 17:530–531, Sept 1974.

[34] Y. Yu, M. Isard, D. Fetterly, M. Budiu, U. Erlingsson, P. K. Gunda, and J. Currey. DryadLINQ: A system for general-purpose distributed data-parallel computing using a high-level language. In OSDI 』08, 2008.

推薦閱讀：