我使用FlinkRunner在本地运行以下Beam管道代码 . PubsubIO用于从主题中读取消息 .

我有一个单独的线程，它定期（每30秒）向主题发布消息，并设置“ts”属性，稍后用于推导事件时间 .

自定义转换转换为KV对 -

private static class PubSubMessageGrouper extends DoFn<PubsubMessage, KV<String, PubsubMessage>> {

    @ProcessElement
    public void processElement(ProcessContext c) {
        PubsubMessage element = c.element();
        KV<String, PubsubMessage> kv = KV.of(element.getAttribute("key"), element);
        c.output(kv);
    }
}

请注意，“key”是发布者线程中较早的消息属性中的键集 . 目的是通过此密钥对消息进行下游分组 .

管道代码 -

PCollection<PubsubMessage> pubsubColl = p
        .apply(PubsubIO.readMessagesWithAttributes()
            .withTimestampAttribute("ts")
            .fromTopic("projects/" + projectName + "/topics/beamtest")
        );


PCollection<KV<String, PubsubMessage>> idfied =
        pubsubColl.apply(ParDo.of(new PubSubMessageGrouper()));

PCollection<KV<String, PubsubMessage>> windowed = idfied
        .apply(Window.<KV<String, PubsubMessage>>into(FixedWindows.of(Duration.standardSeconds(15)))
            .triggering(
                Repeatedly.forever(
                    AfterWatermark.pastEndOfWindow()
                )
            )
            .withAllowedLateness(Duration.standardSeconds(15))
            .discardingFiredPanes());

PCollection<KV<String, Iterable<PubsubMessage>>> grouped = windowed.apply(GroupByKey.create());

grouped.apply(ParDo.of(new KVPrinter()));

变换不是为了便于阅读而链接 . 最后的KVPrinter转换只是打印出从组中收到的消息，一旦我运行，它将随后被实际代码替换 .

当我运行它时，我发现触发器执行了相当长的时间（几分钟或更长时间） . 当它最终触发时，我看到没有收到一些消息（在最后一步） . 这是由于PubsubIO使用的内部水印吗？我的目的是确保所有消息都在groupby中处理，包括在允许的延迟窗口内的后期消息 .

我已经查看了相关问题，但没有得到他们的帮助 . 使用DirectRunner运行相同的操作根本不会产生任何输出，无论我等待多长时间 . 另一点需要注意的是，如果我删除管道的GroupBy部分，则会触发触发器 .

What is the watermark heuristic for PubsubIO running on GCD?

Apache Beam PubSubIO with GroupByKey

Consuming unbounded data in windows with default trigger

注意：即使我没有在Dataflow上测试过，我也用Dataflow标记了这个 . 原因是我打算在Dataflow上部署它，并且还要向正在观看标记的Dataflow团队的人员寻求帮助 .

google-cloud-dataflow apache-flink apache-beam