如何计算 Apache beam 中一个 PCollection 的元素数量

Question

number_items = lines | 'window' >> beam.WindowInto(window.GlobalWindows()) \
    | 'CountGlobally' >> beam.combiners.Count.Globally() \
    | 'print' >> beam.ParDo(PrintFn())

我试着用指纹和日志来显示，但什么也没找到

^{pr2}$

Answer 1

对于批处理，您只需

def print_row(element):
  print element

count_pcol = (
              lines
              | 'Count elements' >> beam.combiners.Count.Globally()
              | 'Print result' >> beam.Map(print_row)
            )

beam.combiners.Count.Globally()是一个PTransform，它使用global combine来计算PCollection的所有元素并生成一个值。在

---

对于流式处理，无法计数元素，因为源是一个无界的pcollection，即它永远不会结束。CombineGlobally在您的情况下，将继续等待输入，而不会产生输出。在

一个可能的解决方案是设置一个窗口函数和一个非默认触发器。在

我编写了一个简单的管道，它将元素划分为20秒的固定窗口，并为每个窗口的每个键计数。您可以根据需要更改窗口和触发器。在

^{pr2}$

Answer 2

我觉得奇怪的是要计算无限集合的元素。我的第一感觉是，永远不要去追全球之窗，因为梁等待着无限的收藏结束。。。只是你在执行一个触发器。在

在文档中挖掘，I found this

Set a non-default trigger. This allows the global window to emit results under other conditions, since the default windowing behavior (waiting for all data to arrive) will never occur

我是对的，有了触发器，结局永远不会发生，它是无限的，无限的。在

你有没有试着跳过窗口直接统计全局？在