检查PCollection是否为空-Apache Beam

Question

有什么方法可以检查PCollection是否为空？

我没有在数据流和Apache Beam的留档中找到任何相关内容。

Answer 1

如果不对其应用PTransform（例如Count.globally（）或Combine.combineFn（）），就无法检查PCollection的大小，因为PCollection不像JavaSDK的典型Collection。

它是有界或无界数据集合的抽象，其中数据被馈送到集合中，以便对其应用操作（例如PTransform）。它也是并行的（正如类开头的P所示）。

因此，您需要一种机制来获取每个工作节点/节点的元素计数，并将其组合以获得值。直到转换结束，才知道它是0还是n。

Answer 2

您没有指定使用哪个SDK，所以我假设使用Python。代码很容易移植到Java。

您可以应用元素的全局计数，然后通过应用简单的比较将数值映射到布尔值。您可以使用pvalue侧向输入该值。AsSingleton函数，如下所示：

import apache_beam as beam
from apache_beam import pvalue

is_empty_check = (your_pcollection
                    | "Count" >> beam.combiners.Count.Globally()
                    | "Is empty?" >> beam.Map(lambda n: n == 0)
                    )

another_pipeline_branch = (
    p
    | beam.Map(do_something, is_empty=pvalue.AsSingleton(is_empty_check))
)

侧面输入的用法如下：

def do_something(element, is_empty):
    if is_empty:
        # yes
    else:
        # no