PyODPS開發中的最佳實踐

pyodps 支援用 python 來對 maxcompute 物件進行操作，它提供了 dataframe api 來用類似 pandas 的介面進行大規模資料分析以及預處理，並且可以用 ml 模組來執行機器學習演算法。

現在為了讓大家能更好地使用 pyodps，我們總結開發過程中的最佳實踐，來讓大家更高效地開發 pyodps 程式。當然，希望大家能一起來幫助我們來完善總結。

我們 pyodps 提供了多種方便拉取資料到本地的操作，因此，很多使用者會試圖把資料拉取到本地處理，然後再上傳到 odps 上。

很多時候，使用者其實根本不清楚這種操作的低效，拉取到本地徹底喪失了 maxcompute 的大規模並行能力。而有的使用者僅僅是需要對單行資料應用乙個 python 函式，或者試圖做一行變多行的操作，這些操作，用 pyodps dataframe 都能輕鬆完成，並且完全利用到了 maxcompute 的平行計算能力。

比如說現在我有乙份資料，都是 json 串，現在我想把 json 串按 key-value 對展開成一行。則可以寫乙個簡單的函式。

in [12]: df
json
0  1  
in [14]: from odps.df import output
in [16]: @output(['k', 'v'], ['string', 'int'])
...: def h(row):
...:     import json
...:     for k, v in json.loads(row.json).items():
...:         yield k, v
...:   
k  v
0  a  1
1  b  2
2  c  4
3  b  3

pyodps dataframe 能夠根據資料**來決定如何執行，比如，通過 pandas dataframe 建立的 pyodps dataframe 則可以使用 pandas 執行本地計算；而使用 maxcompute 表建立的 dataframe 則可以在 maxcompute 上執行。而這兩種方式，除了初始化不同，後續**完全一致，因此，我們可以利用這點來進行本地 debug。

所以我們可以寫出如下的**：

df = o.get_table('movielens_ratings').to_df()
debug = true
if debug:
df = df[:100].to_pandas(wrap=true)

當我們把所有後續**都編寫完成，本地的測試速度就非常快，當測試結束後，我們就可以把 debug 改為 false，這樣後續就能在 odps 上執行全量的計算。

使用本地除錯還有個好處，就是能利用到 ide 的如斷點和單步除錯自定義函式的功能。要知道，在 odps 上執行，是把函式序列化到遠端去執行，所以本地是沒法斷點進入的。而使用本地進行除錯時，則可以斷點進入自定義函式，方便進行除錯。

推薦大家使用 maxcompute studio 來本地除錯 pyodps 程式。

乙個常見的例子就是，計算兩點之間的距離，有多種計算方法，比如歐氏距離、曼哈頓距離等等，我們可以定義一系列函式，在計算時就可以根據具體情況呼叫相應的函式即可。

def euclidean_distance(from_x, from_y, to_x, to_y):
return ((from_x - to_x) ** 2 + (from_y - to_y) ** 2).sqrt()
def manhattan_distance(center_x, center_y, x, y):
return (from_x - to_x).abs() + (from_y - to_y).abs()

呼叫則如下：

in [42]: df from_x from_y to_x to_y 0 0.393094 0.427736 0.463035 0.105007 1 0.629571 0.364047 0.972390 0.081533 2 0.460626 0.530383 0.443177 0.706774 3 0.647776 0.192169 0.244621 0.447979 4 0.846044 0.153819 0.873813 0.257627 5 0.702269 0.363977 0.440960 0.639756 6 0.596976 0.978124 0.669283 0.936233 7 0.376831 0.461660 0.707208 0.216863 8 0.632239 0.519418 0.881574 0.972641 9 0.071466 0.294414 0.012949 0.368514 in [43]: euclidean_distance(df.from_x, df.from_y, df.to_x, df.to_y).rename('distance') distance 0 0.330221 1 0.444229 2 0.177253 3 0.477465 4 0.107458 5 0.379916 6 0.083565 7 0.411187 8 0.517280 9 0.094420 in [44]: manhattan_distance(df.from_x, df.from_y, df.to_x, df.to_y).rename('distance') distance 0 0.392670 1 0.625334 2 0.193841 3 0.658966 4 0.131577 5 0.537088 6 0.114198 7 0.575175 8 0.702558

9 0.132617

乙個常見的需求是，使用者有大概30張表，需要合成一張表，這個時候如果寫 sql，需要寫 union all 30張表，如果表的數量更多，會更讓人崩潰。使用 pyodps，只需要一句話就搞定了。

table_names = ['table1', ..., 'tablen']
dfs = [o.get_table(tn).to_df() for tn in table_names]
reduce(lambda x, y: x.union(y), dfs)

大功告成。稍微解釋下，這裡的 reduce 這句等價於：

df = dfs[0]
for other_df in dfs[1:]:
df = df.union(other_df)

稍微擴充套件下，經常有一些 case 是這樣，使用者要計算的表儲存在某個地方，比如說資料庫，需要根據配置來對錶的字段進行處理，然後對所有表進行 union 或者 join 操作。這個時候，用 sql 實現可能是相當複雜的，但是用 dataframe 進行處理會非常簡單，而實際上我們就有使用者用 pyodps 解決了這樣的問題。

比如上文提到的歐氏距離的計算，實際上，計算的過程都是使用的 dataframe 的內建運算元，比如說指數和 sqrt 等操作，如果我們對一行資料應用自定義函式，則會發現，速度會慢很多。

in [54]: euclidean_distance(df.from_x, df.from_y, df.to_x, df.to_y).rename('distance').mean()
|****************************************==|   1 /  1  (100.00%)         7s
0.5216082314224464
in [55]: @output(['distance'], ['float'])
...: def euclidean_distance2(row):
...:     import math
...:     return math.sqrt((row.from_x - row.to_x) ** 2 + (row.from_y - row.to_y) ** 2)
...: 
|****************************************==|   1 /  1  (100.00%)        27s
0.5216082314224464

可以看到，當我們對一行應用了自定義函式後，執行時間從7秒延長到了27秒，這個資料只是1百萬行資料計算的結果，如果有更大的資料集，更複雜的操作，時間的差距可能會更長。

PyODPS開發中的最佳實踐

PyODPS開發中的最佳實踐

快速開發的最佳實踐整理

基於docker開發的最佳實踐

PyODPS開發中的最佳實踐

PyODPS開發中的最佳實踐

快速開發的最佳實踐整理

基於docker開發的最佳實踐

相關推薦