时间戳

首页 > 解决方案 > 带有 Kafka 源和 Dataflow 运行器的 Beam java SDK 2.10.0：窗口化的 Count.perElement 永远不会触发数据

java - 带有 Kafka 源和 Dataflow 运行器的 Beam java SDK 2.10.0：窗口化的 Count.perElement 永远不会触发数据

问题描述

我在 Google DataFlow 上运行 Beam SDK 到 2.10.0 作业时遇到问题

流程很简单：我使用Kafka作为源，然后应用Fixed windows，然后按键计数。但看起来数据永远不会离开计数阶段，直到工作耗尽。的输出集合Count.PerElement/Combine.perKey(Count)/Combine.GroupedValues.out0始终为零。元素仅在排空 Dataflow 作业后发布。

这是代码：

public KafkaProcessingJob(BaseOptions options) {

    PCollection<GenericRecord> genericRecordPCollection = Pipeline.create(options)
                     .apply("Read binary Kafka messages", KafkaIO.<String, byte[]>read()
                           .withBootstrapServers(options.getBootstrapServers())
                           .updateConsumerProperties(configureConsumerProperties())
                           .withCreateTime(Duration.standardMinutes(1L))
                           .withTopics(inputTopics)
                           .withReadCommitted()
                           .commitOffsetsInFinalize()
                           .withKeyDeserializer(StringDeserializer.class)
                           .withValueDeserializer(ByteArrayDeserializer.class))

                    .apply("Map binary message to Avro GenericRecord", new DecodeBinaryKafkaMessage());

                    .apply("Apply windowing to records", Window.into(FixedWindows.of(Duration.standardMinutes(5)))
                                       .triggering(Repeatedly.forever(AfterWatermark.pastEndOfWindow()))
                                       .discardingFiredPanes()
                                       .withAllowedLateness(Duration.standardMinutes(5)))

                    .apply("Write aggregated data to BigQuery", MapElements.into(TypeDescriptors.strings()).via(rec -> getKey(rec)))
                            .apply(Count.<String>perElement())
                            .apply(
                                new WriteWindowedToBigQuery<>(
                                    project,
                                    dataset,
                                    table,
                                    configureWindowedTableWrite()));   
}

private Map<String, Object> configureConsumerProperties() {
    Map<String, Object> configUpdates = Maps.newHashMap();
    configUpdates.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    return configUpdates;
}

private static String getKey(GenericRecord record) {
    //extract key
}

看起来流永远不会离开舞台.apply(Count.<String>perElement())

有人可以帮忙吗？

标签： javagoogle-cloud-dataflowapache-beam

解决方案

我找到了原因。

它与此处使用的 TimestampPolicy ( .withCreateTime(Duration.standardMinutes(1L))) 有关。

由于我们的 Kafka 主题中存在空分区，主题水印从未使用默认的 TimestampPolicy 进行高级。我需要实施自定义策略来解决问题。

推荐阅读