我们如何实现在 PCollection 中的特定计数后触发的触发器答案

【问题标题】：How can we implement a trigger which fires after certain count in PCollection我们如何实现在 PCollection 中的特定计数后触发的触发器
【发布时间】：2019-02-21 18:39:11
【问题描述】：

我正在编写一个 Beam 数据管道，从 Kafka 等无限源读取数据。我没有执行任何分析功能。我想转换元素并写入接收器，假设在 PCollection 的记录数达到某个阈值之后。这是为了限制发送到接收器的数据

查看了现有的触发器，但无法确定它们是否合适

【问题讨论】：

您是在使用 Beam IO 接收器还是在 DoFn 中推送数据？如果在 DoFn 中，您是否有多个在发送到接收器之前运行 gbk 的密钥？
这听起来像是 AfterPane.elementCountAtLeast(threshold) 的完美用例，它出现在 Beam Programming Guide (beam.apache.org/documentation/programming-guide/…) 的复合触发器部分。

标签： google-cloud-dataflow apache-beam

【解决方案1】：

我已经测试了触发器，它们按预期工作，这是一个scala代码示例

val data: PCollection[Type] = results
  .apply(
  Window
    .into[Type](FixedWindows.of(Duration.millis(2000)))
    .withAllowedLateness(Duration.millis(1000))
    .triggering(AfterPane.elementCountAtLeast(4)
    .accumulatingFiredPanes()
)

它等待 4 个元素然后触发窗口。

【讨论】：