PyODPS 中使用 Python UDF

2018-1-29 14:29:33 2806 UDF

0 摘要： PyODPS 中使用 Python UDF 包含两方面，一个是直接使用，也就是在 MaxCompute SQL 中使用；一个是间接的方式，也就是 PyODPS DataFrame，这种方式你不需要直接写 Python UDF，而是写普通的 Python 函数或者类。点此查看原文：http://click.aliyun.com/m/41092/ PyODPS 中使用 Python UDF 包含两方面，一个是直接使用，也就是在 MaxCompute SQL 中使用；一个是间接的方式，也就是 PyODPS DataFrame，这种方式你不需要直接写 Python UDF，而是写普通的 Python 函数或者类。下面我们分开说明。作为准备工作，我们需要 ODPS 入口，可以通过直接初始化，或者使用 room 机制加载。 from odps import ODPSo = ODPS('your-access-id', 'your-access-key', 'your-project') MaxCompute SQL 中使用 Python UDF 首先，我们需要写一个 Python 文件，假设我们就是把某一列按 csv 格式放的一列转成 json 格式。 import jsonfrom odps.udf import annotate@annotate('string->string')class Transform(object): def evaluate(self, x): columns = list('abc') d = dict(zip(columns, x.split(','))) return json.dumps(d) 假设这个文件叫 my.py，接下来我们就需要创建 py 资源。 r = o.create_resource('csv_to_json.py', 'py', fileobj=open('my.py')) fileobj 参数也可以是 str 类型，就是表示文件的内容接着我们就可以创建 Python UDF 了。 o.create_function('csv_to_json', class_type='csv_to_json.Transform', resources=[r]) 这里我们指定了函数名叫 csv_to_json，主类使我们上传的 csv_to_json.py 文件里的 Transform 类。现在我们就可以在 MaxCompute SQL 中调用这个 UDF 了。 o.execute_sql('select csv_to_json(raw) from pyodps_test_udf') 这样我们就完成了在 PyODPS 中使用 MaxCompute SQL + Python UDF 的整个过程。 PyODPS DataFrame 对于 PyODPS DataFrame 来说，用户只需要写普通的 Python 函数或者类，在函数或者类里，甚至可以读取全局变量，这样给开发带来了极大的方便。和上面的例子目标相同，我们定义一个 transform 函数即可。然后我们对于 DataFrame 的一列调用 map 方法来应用这个函数。 passed_columns = list('abc') # 可以从数据库中读取或者写死def transform(x): import json d = dict(zip(passed_columns, x.split(','))) return json.dumps(d)df.raw.map(transform)In [30]: df raw0 1,2,31 4,5,62 7,8,9In [31]: df.raw.map(transform) raw0 {"a": "1", "c": "3", "b": "2"}1 {"a": "4", "c": "6", "b": "5"}2 {"a": "7", "c": "9", "b": "8"} 实际上，PyODPS DataFrame 在用 MaxCompute 执行的时候，也会创建 Python UDF 来实现这个功能，但用户不需要去创建文件、资源和函数这些过程，一切都是 Python 原生函数和类，整个过程相当顺畅。另外可以看到，在上面的 my.py 里，我们也是定义了一个 columns 参数的，而如果这个参数是通过变量传进去的话，在 Python UDF 里非常麻烦，可能常常需要用一些 tricky 的方法，比如写到某个文件资源，然后在 UDF 里读取之类的。而对于 DataFrame 来说，完全没有这个问题，我们可以自由读取全局变量。不过要注意的是，这个全局变量是被序列化到各个机器上的，所以你修改它不会全局生效。好了，还有什么问题可以随时和我们取得联系。文档：http://pyodps.readthedocs.io/zh_CN/latest/ 代码：https://github.com/aliyun/aliyun-odps-python-sdk ，欢迎提 issue 和 merge request 0
2018-1-29 14:29:33　　评论淘帖0 举报相关推荐 • 如何在Python中使用socket 220 • 如何在Python中使用MQTT 9770 • 如何在Python中使用Selenium 2975 • Python中使用selenium的准备工作 385 • Fluent UDF中文教程 0 • SE5如何在Python中使用SAIL？ 213 • 如何在python代码中使用HTTP代理IP 999 • 如何在python代码中使用HTTP代理IP 1276 • 在PyODPS DataFrame自定义函数中使用pandas、scipy和scikit-learn 2996 • openG中使用python编辑运行报错 3620