将列表转换为 PCollection

问题描述

我目前有一个 DoFn 可以查看存储桶并查看该存储桶和目录前缀中的所有文件。这个 DoFn 返回一个列表而不是一个 PCollection。我如何将此列表转换为 PCollection DoFn 可以使用的 ConvertFileNames？

  # List all the files within a subdir 
  class ListBlobs(beam.DoFn):
    def start_bundle(self):
      self.storage_client = storage.Client()

    def process(self,prefix):
      bucket = self.storage_client.bucket('xxx')
      return list(self.bucket.list_blobs(prefix=prefix))

  # Convert Blobs into filenames as patterns
  class ConvertFileNames(beam.DoFn):
    def process(self,blob):
      return 'gs://' + blob.bucket.name + blob.name

解决方法

如 beam documentation 中所述，Beam DoFn 的处理方法返回可迭代的元素以放置到下游 PCollection 中。所以，在你的例子中，如果我有一个前缀的 PCollection，称之为 prefix_pcoll，那么我可以写

blobs_pcoll = prefix_pcoll | beam.ParDo(ListBlobs())

和 blobs_pcoll 将包含具有此前缀的 blob 列表（即 list(self.bucket.list_blobs(prefix=prefix)) 在所有前缀上的连接）。然后你可以写

converted = blobs_pcoll | beam.ParDo(ConvertFileNames())

你也可以写

converted = blobs_pcoll | beam.Map(
    lambda blob: 'gs://' + blob.bucket.name + blob.name)

您可能还想查看apache_beam.io.fileio.MatchAll。