我在Google DataFlow上有一个问题运行Beam SDK到2.10.0作业
流程很简单:我使用kafka作为源,然后应用固定的窗口,然后按键计数元素。但是看起来数据永远不会离开计数阶段,直到工作排干。Count.PerElement/Combine.perKey(Count)/Combine.GroupedValues.out0
的输出收集始终为零。元素仅在排出数据流工作后发出。
这是代码:
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
}
看起来Flow永远不会离开.apply(Count.<String>perElement())
有人可以帮忙吗?
我找到了原因。
它与此处使用的timestamppolicy有关(.withCreateTime(Duration.standardMinutes(1L))
)。
由于我们的kafka主题中存在空分区,因此使用默认的TimestAmplicy从未提出主题水印。我需要实施自定义策略来解决问题。