SideInput I/O 会影响性能

Posted 2023-03-31

【中文标题】SideInput I/O 会影响性能

【英文标题】：SideInput I/O kills performance

【发布时间】：2020-01-24 13:59:21

【问题描述】：

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

所有这些额外数据都是从 Google Cloud Storage 上的 avro 文件（Dataflow 和 GCS 存储桶使用的同一区域）中读取的，使用 map 函数组织为键值元组，然后使用 pvalue 作为侧输入传递给 DoFn .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 处理方法只包含在侧输入中查找键并根据是否匹配，向元素添加一些额外的数据。

【问题讨论】：

好的，一旦我使用 pvalue.AsList() 将 dict 作为列表传递，似乎会变得更好。根据源代码文档 (beam.apache.org/releases/pydoc/2.4.0/…)，AsList 强制实现侧面输入。似乎 AsDict 没有。有没有办法强制实现字典的实现？为了能够将字典作为列表传递，我将它包装在一个仅包含一个元素的列表中 - 字典。不喜欢这个方案，但它的性能确实好很多。

【参考方案1】：

好的，一个月后再讨论，根据所获得的经验，让我再试一次：

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

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

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

great blog entry "Guide to common Cloud Dataflow use-case patterns" 声明可以将 ParDo 用于流式传输中高达 100 MB 和批处理模式下高达 1 GB 的 SideInput：

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

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

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