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

Question

我们正在使用 Python SDK 在 google 数据流中运行日志文件解析作业。数据分布在数百条每日日志中，我们通过 Cloud Storage 中的文件模式读取这些日志。所有文件的数据量约为 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() 读取原生 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 作为侧输入，那么应该如何构建需要映射的管道，这些映射可以/不应该硬编码到映射函数中？

对我们来说，映射确实经常变化，我需要找到一种方法让“普通”用户提供它。这个想法是在 Cloud Storage 中提供映射 csv，并通过 ReadFromText() 将其简单地合并到管道中。在本地读取它涉及向工作人员提供映射，因此只有技术团队才能做到这一点。

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

上面的所有代码都是用来解释问题的伪代码。对此的任何想法和想法将不胜感激！

Answer 1

对于更有效的侧输入（中小型），您可以使用， beam.pvalue.AsList(mappingTable) 因为AsListBeam 会具体化数据，因此您确定将获得该 pcollection 的内存列表。

旨在用于辅助参数规范——与使用 AsSingleton 和 AsIter 的位置相同，但强制将此 PCollection 物化为列表。

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