无法将 beam python pcollection 转换为列表

Question

TypeError: 'PCollection' object does not support indexing

上述错误是由于尝试将 Pcollection 转换为列表而导致的：

filesList = (files | beam.combiners.ToList())

lines = (p | 'read' >> beam.Create(ReadSHP().ReadSHP(filesList))
            | 'map' >> beam.Map(_to_dictionary))

并且：

def ReadSHP(self, filesList):
    """
    """
    sf = shp.Reader(shp=filesList[1], dbf=filesList[2])

如何解决这个问题？感谢任何帮助。

Answer 1

一般来说，您不能将 PCollection 转换为列表。

PCollection 是一个可能无界且无序的项目集合。 Beam 允许您将转换应用于 PCollection。将 PTransform 应用于 PCollection 会产生另一个 PCollection。并且转换的应用过程可能分布在一组机器上。所以在一般情况下不可能将这样的东西转换成本地内存中的元素集合。

Combiners 只是 PTransforms 的特殊 class。他们所做的是将他们看到的所有元素累加起来，对这些元素应用一些组合逻辑，然后输出组合的结果。例如，组合器可以查看传入的元素，将它们相加，然后将总和作为结果输出。这种组合器将 PCollection 个元素转换为 PCollection 个这些元素的总和。

beam.combiners.ToList 只是应用于 PCollection 的另一个转换，可能在一组工作机器上，并产生另一个 PCollection。但它在产生输出元素之前并没有真正进行任何复杂的组合，它只是将所有可见元素累积到一个列表中，然后输出可见元素列表。因此，它采用键值对（在多台机器上）的元素，将它们放入列表中，然后输出这些列表。

缺少的是从可能的多台机器中获取这些列表并在需要时将它们加载到本地程序中的逻辑。这个问题不容易（如果有的话）以通用方式解决（在所有跑步者之间，所有可能的 IOs 和管道结构）。

其中一个解决方法是向管道添加另一个步骤，将组合输出（例如总和或列表）写入公共存储，例如a table 在某个数据库或文件中。然后当管道完成时，您的程序可以从该位置加载管道执行的结果。

详见文档：

Beam 执行模型：https://beam.apache.org/documentation/execution-model/
设计管道：https://beam.apache.org/documentation/pipelines/design-your-pipeline/
Beam 编程指南：https://beam.apache.org/documentation/programming-guide/

Answer 2

另一种选择是使用 GCE VM 并使用 ogr2ogr 等工具将 shapefile 转换为 GeoJSON。然后可以将 GeoJSON 加载到 BigQuery 中，并可以使用 BigQuery GIS 进行查询。

这是一篇包含更多详细信息的博文
https://medium.com/google-cloud/how-to-load-geographic-data-like-zipcode-boundaries-into-bigquery-25e4be4391c8

无法将 beam python pcollection 转换为列表

Can't convert beam python pcollection into list

python

google-cloud-dataflow

apache-beam