时间戳

首页 > 解决方案 > SideInput I/O 会影响性能

python - SideInput I/O 会影响性能

问题描述

我正在使用 Python SDK 2.15.0 构建数据流管道。在这个管道中，我需要在管道的多个阶段将附加数据加入每个元素。

所有这些额外数据都是从 Google Cloud Storage 上的 avro 文件（用于 Dataflow 和 GCS 存储桶的同一区域）中读取的，使用 map 函数组织为键值元组，然后使用 pvalue.AsDict( ）。侧输入数据在管道执行期间不会改变。

第一次连接（侧输入大小 ~ 1 MB）非常顺利。但是，第二次加入确实表现不佳。它的 sideinput 大小约为 50 MB。

数据流执行图清楚地显示了导致性能不佳的原因：我的 ParDo 步骤消耗的大约 90% 的时间用于读取侧面输入。即使我只使用四个工作节点，从 sideinput 读取的数据量也超过其实际大小几个数量级。

我能做些什么来防止这种情况发生吗？我是否需要以某种方式配置工作缓存大小？在我的 DoFn 的 setup 方法中准备附加数据而不是将其作为 sideinput 传递会更好吗？

这是我准备侧面输入的方法：

sideinput_1 = pvalue.AsDict(p | "Read side input data 1" >> beam.io.ReadFromAvro("gs:/bucket/small_file.avro",0,False,True) \
                              | "Prepare sideinput 1" >> beam.Map(lambda x: (x["KEY"],x["VALUE"])))

# Preparing data for later join
sideinput_2 = pvalue.AsDict(p | "Read side input data 2" >> beam.io.ReadFromAvro("gs://bucket/bigger_file.avro",0,False,True) \
                              | "Prepare side input data 2" >> beam.Map(lambda x: ((x["KEYCOL1"],x["KEYCOL2"],x["KEYCOL3"]),x)))

使用侧输入：


matching = p | "Read address data" >> beam.io.Read(beam.io.BigQuerySource(query=sql_addr, use_standard_sql=True)) \
                 | "Join w/ sideinput1" >> beam.ParDo(Join1(), sideinput_1 ).with_outputs('unmatched', main='matched')                                                                                

result = matching["matched"] | "Join Sideinput 2" >> beam.ParDo(Join2(), sideinput_2 )

DoFn 处理方法仅包含在侧输入中查找键并根据是否存在匹配，向元素添加一些附加数据。

标签： pythonapache-beamdataflow

解决方案

好的，一个月后再讨论，随着经验的积累，让我再试一次：

我很确定侧面输入的性能问题归结为内存交换问题。在管道中，还有一些其他连接非常相似，但侧输入要小得多。他们以合理的时间运行。但是，所有这些连接的比率（IO 字节/侧输入字节）大致相等。

当我将实现从 ParDo with SideInput 切换到CoGroupByKey Transform时，受影响连接的性能提高了几个数量级。

关于侧输入的大小以及何时更喜欢 CoGroupByKey 而不是 DoFn 与 SideInput：

伟大的博客条目“通用云数据流用例模式指南”指出，可以将 ParDo 用于流式传输中高达 100 MB 和批处理模式下高达 1 GB 的 SideInput：

注意：如果可能，将 SideInputs 用于其中一个连接表实际上很小的任何活动 - 在流模式下大约 100MB 或在批处理模式下低于 1GB。这将表现得更好[...]。

我想没有适合每种情况的通用阈值。可能在很大程度上取决于您的管道、机器类型和工人数量等。就我而言，我认为由于管道的高度复杂性，阈值较低。它由大约 40 个转换组成，包括几个连接。

因此，如果您在使用 ParDo 和 Sideinput 进行连接时遇到同样的问题，您可能需要尝试 CoGroupByKey-Transform。

推荐阅读