PyODPS开发中的最佳实践

摘要： PyODPS支持用 Python 来对 MaxCompute 对象进行操作，它提供了 DataFrame API 来用类似 pandas 的接口进行大规模数据分析以及预处理，并且可以用 ml 模块来执行机器学习算法。

点此查看原文：http://click.aliyun.com/m/41091/

PyODPS 支持用 Python 来对 MaxCompute 对象进行操作，它提供了 DataFrame API 来用类似 pandas 的接口进行大规模数据分析以及预处理，并且可以用 ml 模块来执行机器学习算法。

现在为了让大家能更好地使用 PyODPS，我们总结开发过程中的最佳实践，来让大家更高效地开发 PyODPS 程序。当然，希望大家能一起来帮助我们来完善总结。

注：公共云由于未支持 Python UDF，因此本文中提到的自定义函数功能包括 apply 和 map_reduce 等功能公共云用户均暂不可用。

除非数据量很小，否则不要试图进行本地数据处理

我们 PyODPS 提供了多种方便拉取数据到本地的操作，因此，很多用户会试图把数据拉取到本地处理，然后再上传到 ODPS 上。

很多时候，用户其实根本不清楚这种操作的低效，拉取到本地彻底丧失了 MaxCompute 的大规模并行能力。而有的用户仅仅是需要对单行数据应用一个 Python 函数，或者试图做一行变多行的操作，这些操作，用 PyODPS DataFrame 都能轻松完成，并且完全利用到了 MaxCompute 的并行计算能力。

比如说现在我有一份数据，都是 json 串，现在我想把 json 串按 key-value 对展开成一行。则可以写一个简单的函数。

In [12]: dfjson
0  {"a": 1, "b": 2}
1  {"c": 4, "b": 3}In [14]: from odps.df import outputIn [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 计算后端进行高效本地 debug

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)

to_pandas 是将数据下载，根据 wrap 参数来决定是否返回 PyODPS DataFrame，如果是 True，则返回 PyODPS DataFrame；否则，返回 pandas DataFrame。

当我们把所有后续代码都编写完成，本地的测试速度就非常快，当测试结束后，我们就可以把 debug 改为 False，这样后续就能在 ODPS 上执行全量的计算。

使用本地调试还有个好处，就是能利用到 IDE 的如断点和单步调试自定义函数的功能。要知道，在 ODPS 上执行，是把函数序列化到远端去执行，所以本地是没法断点进入的。而使用本地进行调试时，则可以断点进入自定义函数，方便进行调试。

推荐大家使用 MaxCompute studio 来本地调试 PyODPS 程序。

利用 Python 语言特性来实现丰富的功能

编写 Python 函数

一个常见的例子就是，计算两点之间的距离，有多种计算方法，比如欧氏距离、曼哈顿距离等等，我们可以定义一系列函数，在计算时就可以根据具体情况调用相应的函数即可。

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]: dffrom_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.368514In [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.094420In [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 语言的条件和循环语句

一个常见的需求是，用户有大概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.5216082314224464In [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