Apache Beam：使用 Withtimestamp 分配事件时间时出错答案

【问题标题】：Apache Beam: Error assigning event time using WithtimestampApache Beam：使用 Withtimestamp 分配事件时间时出错
【发布时间】：2019-01-25 18:11:50
【问题描述】：

我有一个无限的 Kafka 流发送数据，其中包含以下字段

{"identifier": "xxx", "value": 10.0, "ts":"2019-01-16T10:51:26.326242+0000"}

我使用 apache beam sdk for kafka 读取流

import org.apache.beam.sdk.io.kafka.KafkaIO;
pipeline.apply(KafkaIO.<Long, String>read()
                    .withBootstrapServers("kafka:9092")
                    .withTopic("test")
                    .withKeyDeserializer(LongDeserializer.class)
                    .withValueDeserializer(StringDeserializer.class)
                    .updateConsumerProperties(ImmutableMap.of("enable.auto.commit", "true")) 
                    .updateConsumerProperties(ImmutableMap.of("group.id", "Consumer1"))
                    .commitOffsetsInFinalize()
                    .withoutMetadata()))

由于我想使用 事件时间（在我的示例中为“ts”）进行窗口化，因此我解析传入的字符串并将传入数据流的“ts”字段分配为时间戳。

PCollection<Temperature> tempCollection = p.apply(new SetupKafka())
                    .apply(ParDo.of(new ReadFromTopic()))
                    .apply("ParseTemperature", ParDo.of(new ParseTemperature()));

tempCollection.apply("AssignTimeStamps", WithTimestamps.of(us -> new Instant(us.getTimestamp())));

窗口函数和计算应用如下：

PCollection<Output> output = tempCollection.apply(Window
                .<Temperature>into(FixedWindows.of(Duration.standardSeconds(30)))
                .triggering(AfterWatermark.pastEndOfWindow()
                        .withLateFirings(AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.standardSeconds(10))))
                .withAllowedLateness(Duration.standardDays(1))
                .accumulatingFiredPanes())
                .apply(new ComputeMax());

我从当前 utc 时间延迟 5 秒将数据流式传输到输入流，因为在实际场景中，事件时间戳通常早于处理时间戳。

我收到以下错误：

无法输出时间戳为 2019-01-16T11:15:45.560Z。输出时间戳不能早于当前输入的时间戳 (2019-01-16T11:16:50.640Z) 减去允许的偏差（0 毫秒）。有关更改的详细信息，请参阅 DoFn#getAllowedTimestampSkew() Javadoc 允许的偏差。

如果我将 AssignTimeStamps 的行注释掉，没有错误，但我猜是考虑到处理时间。

如何确保我的计算和窗口基于事件时间而不是处理时间？

请提供一些关于如何处理这种情况的意见。

【问题讨论】：

标签： streaming apache-beam

【解决方案1】：

为了能够使用自定义时间戳，首先你需要实现CustomTimestampPolicy，通过扩展TimestampPolicy<KeyT,ValueT>

例如：

public class CustomFieldTimePolicy extends TimestampPolicy<String, Foo> {


protected Instant currentWatermark;

public CustomFieldTimePolicy(Optional<Instant> previousWatermark) {
    currentWatermark = previousWatermark.orElse(BoundedWindow.TIMESTAMP_MIN_VALUE);
}


@Override
public Instant getTimestampForRecord(PartitionContext ctx, KafkaRecord<String, Foo> record) {
    currentWatermark = new Instant(record.getKV().getValue().getTimestamp());
    return currentWatermark;
}

@Override
public Instant getWatermark(PartitionContext ctx) {
    return currentWatermark;
}

}

然后，当您使用功能接口 TimestampPolicyFactory 设置 KafkaIO 源时，您需要传递您的自定义 TimestampPolicy

KafkaIO.<String, Foo>read().withBootstrapServers("http://localhost:9092")
                .withTopic("foo")
                .withKeyDeserializer(StringDeserializer.class)
                .withValueDeserializerAndCoder(KafkaAvroDeserializer.class, AvroCoder.of(Foo.class)) //if you use avro
                .withTimestampPolicyFactory((tp, previousWatermark) -> new CustomFieldTimePolicy(previousWatermark))
                .updateConsumerProperties(kafkaProperties))

这一行负责创建一个新的 timestampPolicy，传递一个相关的分区和之前的检查点水印见documentation

withTimestampPolicyFactory(tp, previousWatermark) -> new CustomFieldTimePolicy(previousWatermark))

【讨论】：

这个CustomFieldTimePolicy是必须要使用还是我们可以读后映射来分配timestamp，有什么区别？

【解决方案2】：

您有没有机会尝试使用时间戳策略，抱歉我自己没有尝试过这个，但我相信在 2.9.0 中您应该考虑使用该策略以及 KafkaIO 读取。

https://beam.apache.org/releases/javadoc/2.9.0/org/apache/beam/sdk/io/kafka/KafkaIO.Read.html#withTimestampPolicyFactory-org.apache.beam.sdk.io.kafka.TimestampPolicyFactory-

【讨论】：