小编Ral*_*ein的帖子

我们正在使用 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 中提供映射 …