问题描述
限时送ChatGPT账号..我有一个无限制的 Kafka 流发送数据,其中包含以下字段
I have an unbounded Kafka stream sending data with the following fields
{"identifier": "xxx", "value": 10.0, "ts":"2019-01-16T10:51:26.326242+0000"}
我使用用于 kafka 的 apache beam sdk 读取流
I read the stream using the 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.automit", "true"))
.updateConsumerProperties(ImmutableMap.of("group.id", "Consumer1"))
mitOffsetsInFinalize()
.withoutMetadata()))
由于我想使用事件时间(在我的示例中为ts")进行窗口化,因此我解析传入的字符串并将传入数据流的ts"字段分配为时间戳.
Since I want to window using event time ("ts" in my example), i parse the incoming string and assign "ts" field of the incoming datastream as the timestamp.
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 秒,因为在实际场景中,事件时间戳通常早于处理时间戳.
I stream data into the input stream with a lag of 5 seconds from current utc time since in practical scenrios event timestamp is usually earlier than the processing timestamp.
我收到以下错误:
无法输出时间戳为 2019-01-16T11:15:45.560Z.输出时间戳不得早于当前输入的时间戳(2019-01-16T11:16:50.640Z) 减去允许的偏差(0 毫秒).有关更改的详细信息,请参阅 DoFn#getAllowedTimestampSkew() Javadoc允许的偏斜.
Cannot output with timestamp 2019-01-16T11:15:45.560Z. Output timestamps must be no earlier than the timestamp of the current input (2019-01-16T11:16:50.640Z) minus the allowed skew (0 milliseconds). See the DoFn#getAllowedTimestampSkew() Javadoc for details on changing the allowed skew.
如果我将 AssignTimeStamps 的行注释掉,则没有错误,但我猜是在考虑处理时间.
If I comment out the line for AssignTimeStamps, there are no errors but I guess, then it is considering the processing time.
如何确保我的计算和窗口基于事件时间而不是处理时间?
How do I ensure my computation and windows are based on event time and not for processing time?
请提供一些关于如何处理这种情况的意见.
Please provide some inputs on how to handle this scenario.
推荐答案
要能够使用自定义时间戳,首先需要实现CustomTimestampPolicy,通过扩展TimestampPolicy
To be able to use custom timestamp, first You need to implement CustomTimestampPolicy, by extending 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
Then you need to pass your custom TimestampPolicy, when you setting up your KafkaIO source using functional interface TimestampPolicyFactory
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,传递一个相关的分区和之前的检查点水印见文档
This line is responsible for creating a new timestampPolicy, passing a related partition and previous checkpointed watermark see the documentation
withTimestampPolicyFactory(tp, previousWatermark) -> new CustomFieldTimePolicy(previousWatermark))
这篇关于Apache Beam:使用 Withtimestamp 分配事件时间时出错的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!
更多推荐
[db:关键词]
发布评论