我正在使用 Python SDK 2.15.0 构建数据流管道。在此管道中，我需要在管道的多个阶段将附加数据加入到每个元素。
所有这些附加数据都是从 Google Cloud Storage 上的 avro 文件中读取的（Dataflow 和 GCS 存储桶使用的同一区域），使用 map 函数将其组织为键值元组，然后使用 pvalue.AsDict( 作为侧输入传递到 DoFn ）。侧面输入数据在管道执行期间不会改变。
第一次连接（侧面输入大小 ~ 1 MB）进行得非常顺利。然而，第二次连接确实表现不佳。它的 sideinput 大小约为 50 MB。
数据流执行图清楚地显示了导致性能不佳的原因：我的 ParDo 步骤消耗的大约 90% 的时间都花在了读取侧面输入上。即使我只使用四个工作节点，从 sideinput 读取的数据量也超出了其实际大小几个数量级。
我能做些什么来防止这种情况发生吗？我是否需要以某种方式配置工作缓存大小？在我的 DoFn 的设置方法中准备附加数据而不是将其作为 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) \ …