如何使用 Apache Beam Python SDK 使用 ParDo 过滤 PCollection 的元素答案

【问题标题】：How do I Filter elements of a PCollection with a ParDo with Apache Beam Python SDK如何使用 Apache Beam Python SDK 使用 ParDo 过滤 PCollection 的元素
【发布时间】：2018-05-25 22:38:24
【问题描述】：

我有一个 PCollection，我想使用 ParDo 从中过滤掉一些元素。

有什么地方可以找到这个例子吗？

【问题讨论】：

标签： google-cloud-dataflow apache-beam

【解决方案1】：

在 Apache Beam Python SDK 中，有一个过滤器转换，它接收一个 lambda，并过滤掉所有返回 False 的元素。这是一个例子：

filtered_collection = (beam.Create([1, 2, 3, 4, 5])
                       beam.Filter(lambda x: x % 2 == 0))

在这种情况下，filtered_collection 将是一个包含2 和4 的PCollection。

如果您想将此编码为传递给 ParDo 转换的 DoFn，您可以执行以下操作：

class FilteringDoFn(beam.DoFn):
  def process(self, element):
    if element % 2 == 0:
      yield element
    else:
      return  # Return nothing

你可以像这样应用它：

filtered_collection = (beam.Create([1, 2, 3, 4, 5])
                       beam.ParDo(FilteringDoFn()))

与之前一样，filtered_collection 是一个包含2 和4 的PCollection。

【讨论】：