摘要:摘要支持用来对对象进行操作,它提供了来用类似的接口进行大规模数据分析以及预处理,并且可以用模块来执行机器学习算法。现在为了让大家能更好地使用,我们总结开发过程中的最佳实践,来让大家更高效地开发程序。
摘要: 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]: 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 计算后端进行高效本地 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]: 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 语言的条件和循环语句
一个常见的需求是,用户有大概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) ...: In [56]: df.apply(euclidean_distance2, axis=1, reduce=True).mean() |==========================================| 1 / 1 (100.00%) 27s 0.5216082314224464
可以看到,当我们对一行应用了自定义函数后,执行时间从7秒延长到了27秒,这个数据只是1百万行数据计算的结果,如果有更大的数据集,更复杂的操作,时间的差距可能会更长。
总结
利用 PyODPS,我们其实能挖掘更多更灵活、更高效操作 MaxCompute 数据的方式。最佳实践可以不光是我们提供的一些建议,如果你有更多好玩有用的实践,可以多多分享出来。
文档:http://pyodps.readthedocs.io/
代码:https://github.com/aliyun/ali... ,欢迎提 issue 和 merge request
钉钉群:11701793
文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。
转载请注明本文地址:https://www.ucloud.cn/yun/41327.html
摘要:摘要北京云栖大会上阿里云发布了最新的功能,万众期待的功能终于支持啦,我怎么能不一试为快,今天就分享如何通过进行开发。注册函数在脚本中编辑试用好了,一个简单完整的通过开发实践分享完成。 摘要: 2017/12/20 北京云栖大会上阿里云MaxCompute发布了最新的功能Python UDF,万众期待的功能终于支持啦,我怎么能不一试为快,今天就分享如何通过Studio进行Python u...
摘要:中使用首先,我们需要写一个文件,假设我们就是把某一列按格式放的一列转成格式。这里我们指定了函数名叫,主类使我们上传的文件里的类。现在我们就可以在中调用这个了。这样我们就完成了在中使用的整个过程。 摘要: PyODPS 中使用 Python UDF 包含两方面,一个是直接使用,也就是在 MaxCompute SQL 中使用;一个是间接的方式,也就是 PyODPS DataFrame,这种...
摘要:中使用首先,我们需要写一个文件,假设我们就是把某一列按格式放的一列转成格式。这里我们指定了函数名叫,主类使我们上传的文件里的类。现在我们就可以在中调用这个了。这样我们就完成了在中使用的整个过程。 摘要: PyODPS 中使用 Python UDF 包含两方面,一个是直接使用,也就是在 MaxCompute SQL 中使用;一个是间接的方式,也就是 PyODPS DataFrame,这种...
摘要:摘要本文将用户安装时遇到的常见问题列举如下,希望在遇到类似问题时可供借鉴。若不存在,确认此问题。安装后报这通常是由于遇到了依赖问题。上安装时报尝试使用上安装仍然报这是由于系统完整性保护导致的。此后在终端中运行此后再行安装。 摘要: 本文将用户安装 PyODPS 时遇到的常见问题列举如下,希望在遇到类似问题时可供借鉴。在参考下列步骤之前,请首先尝试卸载并重装 PyODPS。卸载的步骤为执...
阅读 3486·2021-11-08 13:20
阅读 3417·2021-09-30 09:48
阅读 2622·2021-09-29 09:41
阅读 629·2021-09-22 15:04
阅读 2563·2021-08-23 09:44
阅读 3759·2020-12-03 17:26
阅读 1058·2019-08-30 14:10
阅读 1620·2019-08-29 18:34