刚开始接触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_list、a 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,除了以下情况:
通过TableReader、table.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 json
filters_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])
)
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的值进行运算,最后输出一个新字段value
res_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 Error
closely_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:
[
"list_value"
],
["list<string>"]
)
def handle_list_type(row):
yield [["test1", "test2"]]
试了很多次才得到这个结果。看到结果的瞬间一下次就想明白了。
用这个例子做个解释:
[
"int_value",
"string_value"
],
["int64", "string"]
)
def handle_list_type(row):
yield 10, "test"
这里的10, "test"是一个元组,恐怕用了list()之类的方法对返回进行了包装。
我最初直接返回["test1", "test2"]的情况下,等同于返回2个string。
所以必须再包一层。想明白了这个原理,那么下面的写法会更加优雅:
[
"list_value"
],
["list<string>"]
)
def handle_list_type(row):
res = ["test1", "test2"]
yield res, #这里有一个逗号
结语
PyODPS的列处理与聚合功能、行处理自定义逻辑,为大数据处理提供了新的视角和工具,让作者乃至更多开发者在告别纯SQL编写的同时,开启了数据处理的新篇章。总之,拥抱变化,勇于实践,PyODPS的潜力等待着每一位数据工程师去挖掘。
团队介绍
我们是淘天集团的场景智能技术团队,作为一支专注于通过AI和3D技术驱动商业创新的技术团队, 依托大淘宝丰富的业务形态和海量的用户、数据, 致力于为消费者提供创新的场景化导购体验, 为商家提供高效的场景化内容创作工具, 为淘宝打造围绕家的场景的第一消费入口。我们不断探索并实践新的技术, 通过持续的技术创新和突破,创新用户导购体验, 提升商家内容生产力, 让用户享受更好的消费体验, 让商家更高效、低成本地经营。
本文分享自微信公众号 - 大淘宝技术(AlibabaMTT)。
如有侵权,请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”,欢迎正在阅读的你也加入,一起分享。