随着执行的推进 + 意外的侧输入行为，数据流管道吞吐量急剧下降

Question

我有一个数据流管道处理大约 1Gb 的数据输入，其中两个 dicts 作为 side_inputs。目标是借助这两个 side_input 从主数据集中计算特征。

管道整体结构如下：

    # First side input, ends up as a 2GB dict with 3.5 million keys
    side_inp1 = ( p | 
        "read side_input1" >> beam.io.ReadFromAvro("$PATH/*.avro") | 
        "to list of tuples" >> beam.Map(lambda row: (row["key"], row["value"]))
    )

    # Second side input, ends up as a 1.6GB dict with 4.5 million keys
    side_inp2 = (p | 
        "read side_input2" >> beam.io.ReadFromAvro("$PATH2/*.avro") | 
        "to list of tuples" >> beam.Map(lambda row: (row["key"], row["value"]))
    )
    
    # The main part of the pipeline, reading an avro dataset of 1 million rows -- 20GB
    (p |
     "read inputs" >> beam.io.ReadFromAvro("$MainPath/*.avro") | 
     "main func" >> beam.Map(MyMapper, pvalue.AsDict(side_inp1), pvalue.AsDict(side_inp2))
     )

这是数据流图：

“特征化”步骤展开：

所以 Featurize 是一个函数，它在侧输入中查找 ids，.gets 向量，并且喜欢 180 种不同的向量点积方法来计算一些特征。这是一个完全受 CPU 限制的进程，预计会比管道的其余部分花费更长的时间，但这里的停顿是很奇怪的事情。

我的问题有两个：

1. 随着数据流管道在流程中进一步移动，数据流管道似乎急剧变慢。我不知道是什么原因以及如何缓解这个问题。可以在下面看到MyMapper 步骤的吞吐量图表，我想知道吞吐量下降（最终从 ~400 行/秒到接近 ~1 行/秒）。

1. side_inputs 的行为对我来说也很奇怪。我希望 side_inputs 只读一次，但是当我查看 Job Metrics / Throughput 图表时，我观察到了以下图表。可以看出，管道不断地在 side_inputs 中读取，而我想要的只是两个保存在内存中的 dicts。

其他作业配置

• 区域：us-central-1a
• machine_type：m1-ultramem-40（40 个 CPU 内核，960GB RAM）
• 磁盘类型/大小：ssd/50GB
• 实验：启用随机服务。
• max_num_workers：1 有助于简化计算和指标，并且不会因自动缩放而变化。

额外观察

• 我经常在 LogViewer 中看到如下日志条目：[INFO] Completed workitem: 4867151000103436312 in 1069.056863785 seconds" 到目前为止，所有已完成的 workItems 都花费了大约 1000-1100 秒，这是另一个令人困惑的原因，为什么吞吐量会下降而处理 workItems 需要与以前相同的时间？并行度是否因某种原因而下降？ （可能是一些我无法控制的隐藏线程阈值，例如 harness_threads？）。

• 在管道的后面部分，查看日志，它看起来执行模式非常顺序（似乎它正在执行 1 个工作项，完成它，转到下一个，这对我来说很奇怪，考虑到有 1TB可用内存和 40 核）

• 有0个错误甚至警告

Answer 1

第 1 点中的吞吐量图表很好地表明您的工作性能有所下降。

侧面输入应为in memory；但是，我不太确定只有 1 个 highmem 节点的管道是一种好方法。由于只有一个节点，管道可能存在难以识别的瓶颈，例如网络或操作系统限制（如在操作系统中打开的max number of files 与加载到内存中的文件相关）。由于梁的架构，我认为即使启用了autoscaling，您也可以拥有更多节点不是问题，因为我们发现自动缩放会自动选择运行您的工作所需的适当数量的工作程序实例。如果您出于其他原因担心计算和指标，请分享。

关于第 2 点，我认为预计会在图表上找到活动，因为侧面输入（在内存中）是正在处理的 read by each element。但是，如果这对您没有意义，您可以随时添加完整的作业图，以便我们了解管道步骤的任何其他详细信息。

我的建议是添加更多工作人员来分配工作负载，因为 PCollection 是一个分布式数据集，将分布在可用节点之间。您可以尝试拥有更多节点的类似计算资源，例如 4 个实例 n2d-highmem-16 (16vCPU 128GB)。通过这种变化，任何瓶颈都可能消失或得到缓解；另外，你可以用同样的方式监控新的工作：

• 请记住在您的管道中使用check errors，这样您就可以识别正在发生/导致性能问题的任何其他问题。

• 在数据流 UI 中检查 CPU and Memory usage。如果在作业级别发生内存错误，Stackdriver 应将它们显示为 memory errors，但还应检查 memory in the host instance 以确保它没有因其他原因达到操作系统中的限制。

• 您可能希望检查 this example 并将侧面输入作为字典。我不是专家，但您可以按照示例中的最佳做法进行操作。

更新

如果机器 n2d-highmem-16 有 OOM，在我看来，每个线束线程都可能使用字典的副本。不太确定配置线程数是否有帮助，但您可以尝试在管道选项中设置number_of_worker_harness_threads。

另一方面，你能扩展步骤Featurize吗？ wall time 在此步骤中非常高（约 6 天），让我们检查吸收这种延迟的 composite transforms。对于有问题的复合变换，让我们知道代码 sn-p。要确定可能出现问题的复合转换，请参阅Side Inputs Metrics，尤其是Time spent writing 和Time spent reading。