数据处理脚手架PyODPS入门体验

原创
09/11 17:52
阅读数 48




本文分享了初次使用PyODPS(Python版的Open Data Processing Service)的心路历程 。作者通过实际案例,深入浅出地探讨了PyODPS相较于传统ODPS SQL在数据处理上的灵活性与便捷性,特别是在处理复杂JSON字段统计与多条件筛选方面展现出的独特优势。同时,文章诚实地指出了PyODPS学习曲线陡峭、运行效率较低及文档细节需完善等不足。借助一系列代码示例,作者不仅揭示了PyODPS中DataFrame操作的精髓,还贴心地总结了调试技巧与最佳实践,为读者搭建起一套实用的数据处理脚手架。

背景介绍

刚开始接触ODPS时,最初有一个需求比较简单,通过ODPS SQL的方式很快得到了解决。


不过最近收到了一个稍微棘手一点的数据处理需求:

  • 统计某些展厅的uv,展厅商品的uv,计算一个比例

  • 统计展厅中某一个json字段内,满足某些条件的数量统计


这里先总结一下PyODPS的优势

  • 灵活的row handle,能灵活地进行数据处理。事实上,需求中也需要对一个json对象进行统计分析,这点上用SQL会非常痛苦。

  • 可以全量加载内容比较少的表、文件资源,降低表处理逻辑上的复杂性。而SQL在这点上没有优势,只能疯狂的join。

  • 优秀的可配置能力,比如说在我这个需求中出现了需要hardCode配置的多关键字过滤

  • 复用SQL处理逻辑,在我的场景里,我需要统计总的比例,与最近15天的比例。但统计逻辑是一样的,不同的是数据的范围~


劣势也很明显:

  • 基本就是写SQL的思路写python。DataFrame基本就是以SQL的表达对数据处理的封装。

  • 运行贼慢,每次调试时间很久。

  • 不能说文档不全面,但是很多语法编译都能过,实际运行没效果。


针对pyodps与python的区别, 我用一段条件判断代码来做个解释:

# 这段是生效的,最后的sql,包含where key in (?) and source == "A"uv_table = visit_table[    visit_table.key.isin(target_key_list) \    & (visit_table.source == "A")].groupby(visit_table.target_id)# 这段是有问题的。最后的sql,只有where source == "A"uv_table = visit_table[    visit_table.key.isin(target_key_list)    and (visit_table.source == "A")].groupby(visit_table.target_id)# 这段也是ok的,看起来是官方文档中推荐的写法uv_table = visit_table[    visit_table.key.isin(target_key_list) & (visit_table.source == "A")].groupby(visit_table.target_id)# 这段也是ok的,这里换行没有任何问题,也就是说\可加可不加。uv_table = visit_table[    visit_table.key.isin(target_key_list)    & (visit_table.source == "A")].groupby(visit_table.target_id)# 这段就是不行的,会丢失in。对应到SQL就是 where true and source == "A"uv_table = visit_table[    visit_table.key in target_key_list & (visit_table.source == "A")].groupby(visit_table.target_id)


上面的代码示例,全部都可以正常编译且执行,但是从结果上来说却大有不同:

  • 在PyOdps对象中,使用了python语言特性的判断条件,如a in a_lista is None这类逻辑均不会生效,会被忽略。取而代之,应该使用a.isin(a_list)a.isnull()这样的pyodps方法。


所以先解释下为啥拿判断条件开头:已经被坑了n次了,编译全过,运行完成,但结果却经常没生效某一些条件,导致来来回回全文检查。甚至我感觉这个是目前来说最容易踩坑的点。


最后推荐的判断条件写法如下:

uv_table = visit_table[    (visit_table.key.isin(target_key_list))    & (visit_table.source == "A")].groupby(visit_table.target_id)

每个判断条件均用()包裹,并换行or不换行&与、|或、~非,分割条件。


从这个点延伸开,我们已经发现了,PyODPS中,有两种思路。一种是面向DataFrame而另一种则是面向纯Python


正常来说通篇均为面向DataFrame,除了以下情况:

  • 通过TableReadertable.head(10)等方法将表数据读取为python的list对象数据。后续的处理逻辑均需要用python去解决。

  • @output代码处理逻辑,全部为python的能力去解决。

  • class Agg,这种自定义聚合代码,均为Python的逻辑进行处理。


而所有与DataFrame相关的逻辑,都必须查文档来处理,比如说对json的处理,我们就需要使用df.func.get_json_obj(table_name.field),而不能使用python的json.loads()

数据的空判断则需要用a.isnull()或者a.notnull()等方法。


pyodps文档:https://pyodps.readthedocs.io/zh-cn/stable/api-df.html


写完了脚本回来一看就有种理所当然的感觉~不得不说设计上还是巧妙的


但是这里不得不提一个点:

  • PyODPS如果用了错误的方式调用,则也不会错误,必须仔细检查我们的SQL。是否达到我们预期的想法。


所以调试我们的PyODPS,就是重中之重


同时,对于去重来说,官方文档的方法好像是有问题的。


# 这段只会提示no field in XXXXGroupBySequence(具体是啥记不住了)show_room_uv = show_room_uv.agg(show_room_uv=show_room_uv.visitor_id.unique())
# 反复验证后,正确的去重计数是nunique()

吐槽结束,接下来开始本期的重点。


PyODPS开发的基本脚手架


咱们的这个数据处理的功能非常适合以一个基础的脚手架起步~
这里我根据自己的开发经验总结了一个:


from odps.df import DataFrame, Scalar, func, output
# args也是一个内置对象,就是我们在调度配置中的参数bizdate = args["bizdate"]
output_table = "xxxx"# 加载我们的数据表。o是一个内置对象。也有o.get_table("xxx").to_df()的写法。data_process_table = DataFrame(o.get_table("xxxx"))
# 加载我们的数据import jsonfilters_words = []# filters_words.txt就是我们放在MaxCompute -> 资源下的文件。with o.get_resource('filters_words.txt').open('r', encoding='utf-8') as f: filters_words = json.loads(f.read())# 这里就是odps语法了,这里等同于 where content in (?, ?)。包括说content is null,就是content.isnull()。# 在DataFrame的范围内,需要遵从官方的API。data_process_table[ data_process_table.content.isin(filters_words)]# 如果要like怎么处理呢?data_process_table = data_process_table.query( " or ".join([f"content.contains('{x}')" for x in filters_words]))
@output(["content_len"], ["int64"])def handle(row): # 这里是按行处理数据。如果要做reduce之类的多行处理,要通过agg自定义聚合的逻辑。 # python的数据处理 yield len(row.content)
# 很有意思的列处理,这个操作相当于handle处理完多了一列content_len。# 另外我们可以理解每次[]处理完之后,是拿到了一个新的DataFrame对象。res_t = data_process_table[ data_process_table, data_process_table.apply(handle, axis = 1)]# 这一部分是后补的,纯手撕代码。class Agg(object): def buffer(self): # 定义你心仪的聚合结果对象。自定义聚合的本质就是将结果加到这个buffer对象里 return { "merge_length": 0 } def __call__(self, buffer, content_len): if content_len is not None: # 当前聚合对象数据合并。数据被分成了无数个小片,这是其中一片的n条数据聚合 buffer["merge_length"] += content_len
def merge(self, buffer, pbuffer): # 和其他的聚合对象进行合并~ buffer["merge_length"] += pbuffer["merge_length"]
def getvalue(self, buffer): return buffer["merge_length"]# 调用聚合方法to_agg = agg( [ # output输出的新字段,我们作为聚合的value去处理 res_t.content_len ], Agg, rtype="int64", )# 用id去做聚合,对content_len的值进行运算,最后输出一个新字段valueres_t = res_t.groupby("id").agg(value=to_agg)# 此时res_t的列有 id、value,两个字段。# 调试用,看看数据,最后换成persist持久化到output表里。res_t.head(10)# 最后要写数据库了,直接用下面的方法.# res_t.persist(output_table, partition=f"ds='{bizdate}'", drop_partition=True, create_partition=True)

在总结脚手架的时候,不得不说PyODPS是一个精妙的设计,估计是再也回不去写SQL的日子了。


PyODPS核心思想就两点:

  • 在DataFrame中做列处理和聚合。删除列,按条件过滤,整列计算。

  • 在handle中做行处理,同时定义按行处理后的输出列。难以分析的学习,直接用代码分析~


核心文档,写的过程中还是需要不断借鉴:

  • 列运算

  • 聚合操作,里面的unique应该是过期了,用nunique。一旦聚合后就是一个GroupBy对象,需要调用agg对聚合结果处理后,回到DataFrame


另外还得吐槽一句,确实很难写。


# 看着是不是没问题。直接报错.agg Syntax Errorclosely_count_table = data_process_table.groupby('content_len').agg(closely_count = data_process_table.content_len)

这个写法里有两个问题:

  • 第一个是我怎么都摸不清的换行问题,即使不是这个情况,有的时候换行就会解析不了,包括条件判断。

  • 第二个呢,就是对象问题,agg函数的入参应该是一个GroupBy对象,而不是DataFrame。


但是,自定义聚合连着写又没啥问题,只能说最终解释权都在PyOdps。所以这里这么写是最保险的。


closely_count_table = data_process_table.groupby('content_len')closely_count_table = closely_count_table.agg(closely_count = closely_count_table.content_len)

即使同为DataFrame也有一样的问题,不要妄想用多个[][]来完成多次处理。第一个[]内可以用当前的DataFrame,但第二个[]就不一样了,它需要的是第一个[]返回的DataFrame对象。举个例子:


# 过滤了content_len小于等于10的数据,并输出content.# 但这个大概率是错的。虽然我没试过。data_process_table = data_process_table[    data_process_table.content_len > 10][    data_process_table.content]# 你可以这么写,因为过滤content_len的DataFrame仍然有content字段,通过字符是可以取出来的。data_process_table = data_process_table[    data_process_table.content_len > 10]["content"]# 保守写法data_process_table = data_process_table[data_process_table.content_len > 10]data_process_table = data_process_table[data_process_table.content]

关于list type:

@output(    [        "list_value"    ],    ["list<string>"])def handle_list_type(row):    yield [["test1", "test2"]]

试了很多次才得到这个结果。看到结果的瞬间一下次就想明白了。

用这个例子做个解释:

@output(    [        "int_value",        "string_value"    ],    ["int64", "string"])def handle_list_type(row):    yield 10, "test"

这里的10, "test"是一个元组,恐怕用了list()之类的方法对返回进行了包装。


我最初直接返回["test1", "test2"]的情况下,等同于返回2个string


所以必须再包一层。想明白了这个原理,那么下面的写法会更加优雅:

@output(    [        "list_value"    ],    ["list<string>"])def handle_list_type(row):    res = ["test1", "test2"]    yield res, #这里有一个逗号

结语

PyODPS的列处理与聚合功能、行处理自定义逻辑,为大数据处理提供了新的视角和工具,让作者乃至更多开发者在告别纯SQL编写的同时,开启了数据处理的新篇章。总之,拥抱变化,勇于实践,PyODPS的潜力等待着每一位数据工程师去挖掘。


团队介绍


我们是淘天集团的场景智能技术团队,作为一支专注于通过AI和3D技术驱动商业创新的技术团队, 依托大淘宝丰富的业务形态和海量的用户、数据, 致力于为消费者提供创新的场景化导购体验, 为商家提供高效的场景化内容创作工具, 为淘宝打造围绕家的场景的第一消费入口。我们不断探索并实践新的技术, 通过持续的技术创新和突破,创新用户导购体验, 提升商家内容生产力, 让用户享受更好的消费体验, 让商家更高效、低成本地经营。




¤  拓展阅读  ¤

3DXR技术 |  终端技术 |  音视频技术

服务端技术 | 技术质量 | 数据算法






本文分享自微信公众号 - 大淘宝技术(AlibabaMTT)。
如有侵权,请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”,欢迎正在阅读的你也加入,一起分享。

展开阅读全文
加载中
点击引领话题📣 发布并加入讨论🔥
0 评论
0 收藏
0
分享
返回顶部
顶部