PyODPS 中使用 Python UDF

发表时间:2018-02-28

摘要: 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 机制 加载。

fromodps importODPS o = ODPS( 'your-access-id', 'your-access-key', 'your-project')

MaxCompute SQL 中使用 Python UDF

首先,我们需要写一个 Python 文件,假设我们就是把某一列按 csv 格式放的一列转成 json 格式。

importjson fromodps.udf importannotate @annotate('string->string')classTransform(object):defevaluate(self, x):columns = list( 'abc') d = dict(zip(columns, x.split( ','))) returnjson.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(' selectcsv_to_json(raw) frompyodps_test_udf')

这样我们就完成了在 PyODPS 中使用 MaxCompute SQL + Python UDF 的整个过程。

PyODPS DataFrame

对于 PyODPS DataFrame 来说,用户只需要写普通的 Python 函数或者类,在函数或者类里,甚至可以读取全局变量,这样给开发带来了极大的方便。

和上面的例子目标相同,我们定义一个 transform 函数即可。然后我们对于 DataFrame 的一列调用 map 方法来应用这个函数。

passed_columns = list( 'abc') # 可以从数据库中读取或者写死deftransform(x):importjson d = dict(zip(passed_columns, x.split( ','))) returnjson.dumps(d) df.raw.map(transform) In[ 30]: df raw 01, 2, 314, 5, 627, 8, 9In[ 31]: df .raw.map(transform) raw 0{ "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