因此，我正在評(píng)估 Kafka Streams 以及它可以做些什么來(lái)查看它是否適合我的用例，因?yàn)槲倚枰扛?15 分鐘、每小時(shí)、每天聚合一次傳感器的數(shù)據(jù)，并且由于它的 Windowing 功能而發(fā)現(xiàn)它很有用。因?yàn)槲铱梢酝ㄟ^(guò)應(yīng)用創(chuàng)建窗口，windowedBy()但KGroupedStream問(wèn)題是窗口是在 UTC 中創(chuàng)建的，我希望我的數(shù)據(jù)按其原始時(shí)區(qū)而不是按 UTC 時(shí)區(qū)分組，因?yàn)樗璧K了聚合，所以任何人都可以幫助我解決這個(gè)問(wèn)題。
2 回答
明月笑刀無(wú)情
TA貢獻(xiàn)1828條經(jīng)驗(yàn) 獲得超4個(gè)贊
您可以使用自定義“移動(dòng)”時(shí)間戳TimestampExtractor- 在將結(jié)果寫(xiě)回輸出主題之前，您可以使用 aTransformer并通過(guò)context.forward(key, value, To.all().withTimestamps()).
功能請(qǐng)求票：https ://issues.apache.org/jira/browse/KAFKA-7911
侃侃無(wú)極
TA貢獻(xiàn)2051條經(jīng)驗(yàn) 獲得超10個(gè)贊
因此，為了解決這個(gè)問(wèn)題，我創(chuàng)建了自定義TimestampExtractor并使用它來(lái)更改流窗口創(chuàng)建時(shí)間以記錄來(lái)自有效負(fù)載的時(shí)間，如下所示。
public class RecordTimeStampExtractor implements TimestampExtractor {
@Override
public long extract(ConsumerRecord<Object, Object> record, long previousTimestamp) {
JsonObject data = (JsonObject) new JsonParser().parse(record.value().toString());
Timestamp recordTimestamp = Timestamp.valueOf(data.get(Constant.SLOT).getAsString());
return recordTimestamp.getTime();
}
}
所以現(xiàn)在我已經(jīng)用我的本地時(shí)區(qū)測(cè)試了它，因?yàn)樽蛱焓?IST 05:30，它的工作正常，kafka 流也正在根據(jù)記錄時(shí)間戳創(chuàng)建窗口。也將使用其他時(shí)區(qū)進(jìn)行測(cè)試并更新答案
添加回答
舉報(bào)
0/150
提交
取消