摘要：摘要支持用來對(duì)對(duì)象進(jìn)行操作，它提供了來用類似的接口進(jìn)行大規(guī)模數(shù)據(jù)分析以及預(yù)處理，并且可以用模塊來執(zhí)行機(jī)器學(xué)習(xí)算法。現(xiàn)在為了讓大家能更好地使用，我們總結(jié)開發(fā)過程中的最佳實(shí)踐，來讓大家更高效地開發(fā)程序。

摘要： PyODPS支持用 Python 來對(duì) MaxCompute 對(duì)象進(jìn)行操作，它提供了 DataFrame API 來用類似 pandas 的接口進(jìn)行大規(guī)模數(shù)據(jù)分析以及預(yù)處理，并且可以用 ml 模塊來執(zhí)行機(jī)器學(xué)習(xí)算法。

點(diǎn)此查看原文：http://click.aliyun.com/m/41091/

PyODPS 支持用 Python 來對(duì) MaxCompute 對(duì)象進(jìn)行操作，它提供了 DataFrame API 來用類似 pandas 的接口進(jìn)行大規(guī)模數(shù)據(jù)分析以及預(yù)處理，并且可以用 ml 模塊來執(zhí)行機(jī)器學(xué)習(xí)算法。

現(xiàn)在為了讓大家能更好地使用 PyODPS，我們總結(jié)開發(fā)過程中的最佳實(shí)踐，來讓大家更高效地開發(fā) PyODPS 程序。當(dāng)然，希望大家能一起來幫助我們來完善總結(jié)。

注：公共云由于未支持 Python UDF，因此本文中提到的自定義函數(shù)功能包括 apply 和 map_reduce 等功能公共云用戶均暫不可用。

除非數(shù)據(jù)量很小，否則不要試圖進(jìn)行本地?cái)?shù)據(jù)處理

我們 PyODPS 提供了多種方便拉取數(shù)據(jù)到本地的操作，因此，很多用戶會(huì)試圖把數(shù)據(jù)拉取到本地處理，然后再上傳到 ODPS 上。

很多時(shí)候，用戶其實(shí)根本不清楚這種操作的低效，拉取到本地徹底喪失了 MaxCompute 的大規(guī)模并行能力。而有的用戶僅僅是需要對(duì)單行數(shù)據(jù)應(yīng)用一個(gè) Python 函數(shù)，或者試圖做一行變多行的操作，這些操作，用 PyODPS DataFrame 都能輕松完成，并且完全利用到了 MaxCompute 的并行計(jì)算能力。

比如說現(xiàn)在我有一份數(shù)據(jù)，都是 json 串，現(xiàn)在我想把 json 串按 key-value 對(duì)展開成一行。則可以寫一個(gè)簡(jiǎn)單的函數(shù)。

In [12]: df
               json
0  {"a": 1, "b": 2}
1  {"c": 4, "b": 3}

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
    ...:   

In [21]: df.apply(h, axis=1)
   k  v
0  a  1
1  b  2
2  c  4
3  b  3

而這些操作，幾乎全部都可以用 apply（axis=1）和 map_reduce 接口完成。

使用 pandas 計(jì)算后端進(jìn)行高效本地 debug

PyODPS DataFrame 能夠根據(jù)數(shù)據(jù)來源來決定如何執(zhí)行，比如，通過 pandas DataFrame 創(chuàng)建的 PyODPS DataFrame 則可以使用 pandas 執(zhí)行本地計(jì)算；而使用 MaxCompute 表創(chuàng)建的 DataFrame 則可以在 MaxCompute 上執(zhí)行。 而這兩種方式，除了初始化不同，后續(xù)代碼完全一致，因此，我們可以利用這點(diǎn)來進(jìn)行本地 debug。

所以我們可以寫出如下的代碼：

df = o.get_table("movielens_ratings").to_df()
DEBUG = True
if DEBUG:
    df = df[:100].to_pandas(wrap=True)

to_pandas 是將數(shù)據(jù)下載，根據(jù) wrap 參數(shù)來決定是否返回 PyODPS DataFrame，如果是 True，則返回 PyODPS DataFrame；否則，返回 pandas DataFrame。

當(dāng)我們把所有后續(xù)代碼都編寫完成，本地的測(cè)試速度就非常快，當(dāng)測(cè)試結(jié)束后，我們就可以把 debug 改為 False，這樣后續(xù)就能在 ODPS 上執(zhí)行全量的計(jì)算。

使用本地調(diào)試還有個(gè)好處，就是能利用到 IDE 的如斷點(diǎn)和單步調(diào)試自定義函數(shù)的功能。要知道，在 ODPS 上執(zhí)行，是把函數(shù)序列化到遠(yuǎn)端去執(zhí)行，所以本地是沒法斷點(diǎn)進(jìn)入的。而使用本地進(jìn)行調(diào)試時(shí)，則可以斷點(diǎn)進(jìn)入自定義函數(shù)，方便進(jìn)行調(diào)試。

推薦大家使用 MaxCompute studio 來本地調(diào)試 PyODPS 程序。

利用 Python 語言特性來實(shí)現(xiàn)豐富的功能

編寫 Python 函數(shù)

一個(gè)常見的例子就是，計(jì)算兩點(diǎn)之間的距離，有多種計(jì)算方法，比如歐氏距離、曼哈頓距離等等，我們可以定義一系列函數(shù)，在計(jì)算時(shí)就可以根據(jù)具體情況調(diào)用相應(yīng)的函數(shù)即可。

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()

調(diào)用則如下：

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

利用 Python 語言的條件和循環(huán)語句

一個(gè)常見的需求是，用戶有大概30張表，需要合成一張表，這個(gè)時(shí)候如果寫 SQL，需要寫 union all 30張表，如果表的數(shù)量更多，會(huì)更讓人崩潰。使用 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 這句等價(jià)于：

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

稍微擴(kuò)展下，經(jīng)常有一些 case 是這樣，用戶要計(jì)算的表保存在某個(gè)地方，比如說數(shù)據(jù)庫，需要根據(jù)配置來對(duì)表的字段進(jìn)行處理，然后對(duì)所有表進(jìn)行 union 或者 join 操作。這個(gè)時(shí)候，用 SQL 實(shí)現(xiàn)可能是相當(dāng)復(fù)雜的，但是用 DataFrame 進(jìn)行處理會(huì)非常簡(jiǎn)單，而實(shí)際上我們就有用戶用 PyODPS 解決了這樣的問題。

盡量使用內(nèi)建算子，而不是自定義函數(shù)

比如上文提到的歐氏距離的計(jì)算，實(shí)際上，計(jì)算的過程都是使用的 DataFrame 的內(nèi)建算子，比如說指數(shù)和 sqrt 等操作，如果我們對(duì)一行數(shù)據(jù)應(yīng)用自定義函數(shù)，則會(huì)發(fā)現(xiàn)，速度會(huì)慢很多。

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)
    ...: 

In [56]: df.apply(euclidean_distance2, axis=1, reduce=True).mean()
|==========================================|   1 /  1  (100.00%)        27s
0.5216082314224464

可以看到，當(dāng)我們對(duì)一行應(yīng)用了自定義函數(shù)后，執(zhí)行時(shí)間從7秒延長(zhǎng)到了27秒，這個(gè)數(shù)據(jù)只是1百萬行數(shù)據(jù)計(jì)算的結(jié)果，如果有更大的數(shù)據(jù)集，更復(fù)雜的操作，時(shí)間的差距可能會(huì)更長(zhǎng)。

總結(jié)

利用 PyODPS，我們其實(shí)能挖掘更多更靈活、更高效操作 MaxCompute 數(shù)據(jù)的方式。最佳實(shí)踐可以不光是我們提供的一些建議，如果你有更多好玩有用的實(shí)踐，可以多多分享出來。

文檔：http://pyodps.readthedocs.io/
代碼：https://github.com/aliyun/ali... ，歡迎提 issue 和 merge request
釘釘群：11701793