python - Google Dataflow/Apache Beam Python - 来自 PCollection 的侧输入会降低性能

Question

我们正在使用 Python SDK 在谷歌数据流中运行日志文件解析作业。数据分布在数百个日常日志中，我们通过云存储中的文件模式读取这些日志。所有文件的数据量约为 5-8 GB(gz 文件)，总共 50-80 百万行。

loglines = p | ReadFromText('gs://logfile-location/logs*-20180101')

此外，我们有一个简单的(小的)映射 csv，它将日志文件条目映射到人类可读的文本。大约有 400 行，大小为 5 kb。

例如，带有 [param=testing2] 的日志文件条目应映射到最终输出中的“客户请求 14 天免费产品试用”。

我们在带有 sideinput 的简单 beam.Map 中执行此操作，如下所示:

customerActions = loglines | beam.Map(map_logentries,mappingTable)

其中map_logentries为映射函数，mappingTable为映射表。

然而，这只有在我们通过 open()/read() 读取 native python 中的映射表时才有效。如果我们通过 ReadFromText() 使用光束管道执行相同的操作，并将生成的 PCollection 作为侧输入传递给 Map，如下所示:

mappingTable = p | ReadFromText('gs://side-inputs/category-mapping.csv')    
customerActions = loglines | beam.Map(map_logentries,beam.pvalue.AsIter(mappingTable))

性能完全下降到每秒 2-3 个项目。

现在，我的问题:

1. 为什么性能会如此糟糕，通过一个PCollection 作为辅助输入？
2. 如果可能不建议使用PCollections 作为辅助输入，应该如何构建诸如需要可以/不应该硬编码的映射的管道映射函数？

对于我们来说，映射确实经常变化，我需要找到一种方法让“普通”用户提供它。我们的想法是让映射 csv 在云存储中可用，并通过 ReadFromText() 将其简单地合并到管道中。在本地读取它涉及向工作人员提供映射，因此只有技术团队可以执行此操作。

我知道边输入存在缓存问题，但这肯定不适用于 5kb 输入。

以上代码均为伪代码，用于说明问题。对此有任何想法和想法，我们将不胜感激!

Answer 1

对于更高效的侧输入(中小型)，您可以利用
beam.pvalue.AsList(映射表)
因为 AsList 导致 Beam 实现数据，所以您确定您将获得该 pcollection 的内存列表。

Intended for use in side-argument specification---the same places where AsSingleton and AsIter are used, but forces materialization of this PCollection as a list.

Source: https://beam.apache.org/documentation/sdks/pydoc/2.2.0/apache_beam.pvalue.html?highlight=aslist#apache_beam.pvalue.AsList