Python+大數據運算平台，PyODPS架構搭建

資料分析與機器學習

大數據基本上都是建立在Hadoop系統的生態上的，其實一個Java的環境。很多人喜歡用Python和R來進行資料分析，但這往往對應一些小數據的問題，或是本地資料處理的問題。如何將二者進行結合使其具有更大的價值?Hadoop現有的生態系和現有的Python環境如上圖所示。

MaxCompute

MaxCompute是針對離線運算的大數據平台，提供TB/PB級的資料處理，多租用戶、開箱即用、隔離機制確保安全。 MaxCompute上主要分析的工具就是SQL，SQL非常簡單、容易上手，屬於描述型。 Tunnel提供資料上傳下載通道，不需要經過SQL引擎的調度。

Pandas

Pandas是基於numpy的資料分析的工具，裡面最重要的結構是DataFrame，提供一系列繪圖的API，背後是matplotlib的操作，非常容易和Python第三方函式庫互動。

PyODPS架構

PyODPS即利用Python進行大數據分析，其架構如上圖所示。底層是基礎API，可以利用其操作MaxCompute上的表、函數或資源。再來是DataFrame框架，DataFrame包含兩個部分，一部分是前端，定義了一套表達式的操作，使用者寫的程式碼會轉換成表達式樹，這與普通的語言是一樣的。使用者可以自訂函數，也可以進行視覺化，與第三方函式庫互動。後端最下面是Optimizer，其作用是對表達式樹進行最佳化。 ODPS和pandas都是透過compiler和analyzer提交到Engine來執行。

背景

為什麼要做DataFrame框架?

對於任何一個大數據分析工具，都會面臨三個維度上的問題：表達力，API、語法、程式語言是否簡單、符合直覺式數據，儲存、元資料是否能壓縮、有效?引擎，計算的性能是否足夠?所以就會面臨pandas和SQL兩個選擇。

如上圖所示，pandas的表達力非常好，但是其數據只能放在內存中，引擎是單機的，受限於本機的性能。 SQL的表達力有限，但是可以用於大量的數據，數據量小的時候沒有引擎的優勢，數據量大的時候引擎會變得很有優勢。 ODPS的目標是綜合這兩者的優點。

PyODPS DataFrame

PyODPS DataFrame是使用Python語言寫的，可以使用Python的變數、條件判斷、循環。可以使用pandas類似的語法，定義了自己的一套前端，有了更好的表達力。後端可以根據資料來源來決定具體執行的引擎，是visitor的設計模式，可擴充。整個執行是延遲執行，除非使用者呼叫立即執行的方法，否則是不會直接執行的。

從上圖可以看出，文法非常類似pandas。

表達式與抽象語法樹

從上圖可以看出，使用者從一個原始的Collection來進行GroupBy操作，再進行列選擇的操作，最下面是Source的Collection。取了兩個欄位species，這兩個欄位是做By操作的，pental_length是進行聚合的操作取聚合值。 Species欄位是直接取出來，shortest欄位是進行加一的操作。

Optimizer(操作合併)

後端首先會使用Optimizer對表達式樹進行優化，先做GroupBy，然後在上面做列選擇，透過操作合併可以去除petal_length做聚合操作，再加一，最後形成了GroupBy的Collection。

Optimizer(列剪枝)

用戶join了兩個data frame，再取來自data frame 的兩個列的時候，如果提交到一個大數據的環境，這樣一個過程是非常低下的，因為不是每個列都用到了。所以要對joined下的列進行剪枝操作。例如，data frame1我們只用到了其中的一個字段，我們只需要將字段截取出來做一個projection來形成新的Collection，data frame2也類似。這樣，對這兩部分進行校驗操作的時候就能極大的減少資料的輸出量。

Optimizer(謂詞下推)

如果對兩個data frame進行joined然後再分別進行過濾的話，這個過濾操作是應該下推到下面來執行的，這樣就能減少joined 的輸入的量。

視覺化

提供了visualize()來方便使用者進行視覺化。在右邊的範例可以看到，ODSP SQL後端會compile成一條SQL執行。

後端

從上圖可以看出，計算後端是非常靈活的。使用者甚至可以joined一個pandas的data frame和maxcompute上一個表格的資料。

Analyzer

Analyzer的作用是針對特定的後端，將一些操作轉換。例如：

有些操作例如value_counts，pandas本身支持，因此對於pandas後端，無需處理;對於ODPS SQL後端，沒有一個直接的操作來執行，所以在analyzer執行的時候，會被改寫成groupby + sort的操作;

還有一些算子，在compile到ODPS SQL時，沒有內建函數能完成，會被改寫成自訂函數。

ODPS SQL後端

ODPS SQL後端怎麼進行SQL編譯再執行的操作?編譯器可以從上到下遍歷表達式樹，找到Join或Union。對於子過程，進行遞歸compile。再到Engine來具體執行時，會使用Analyzer對表達式樹進行改寫，compile自上而下的子過程，自底向上compile成SQL子句，最終得到完整的SQL語句，提交SQL並返回任務。

pandas後端

首先訪問這個表達式樹，然後對每個表達式樹節點對應到pandas操作，整個表達式樹遍歷完之後就會形成DAG。 Engine執行依照DAG拓樸順序執行，不斷地把它應用到pandas操作，最後得到一個結果。對於大數據環境來說，pandas後端的作用是做本地DEBUG;當資料量很小時，我們可以使用pandas來計算。

難點+坑

後端編譯出錯容易丟失上下文，多次optimize和analyze，導致難以查出是之前哪處visit node導致。解決：確保每個模組獨⽴立性、測試完備;

bytecode相容問題，maxcompute只支援Python2.7的自訂函數的執行;

SQL的執行順序。

ML機器學習

機器學習是輸入輸出一個data frame。例如，有一個iris的data frame，先用name字段來做一個分類字段，呼叫split方法將其分成60%的訓練資料和40%的測試資料。接著初始化一個RandomForests，裡面有一棵決策樹，呼叫train方法訓練訓練數據，呼叫predict方法形成一個預測數據，呼叫segments[0]就可以看到可視化結果。

未來計畫

分散式numpy，DataFrame基於分散式numpy的後端;

記憶體計算，提升互動式體驗;

Tensorflow。