第七色在线视频,2021少妇久久久久久久久久,亚洲欧洲精品成人久久av18,亚洲国产精品特色大片观看完整版,孙宇晨将参加特朗普的晚宴

為了賬號安全,請及時綁定郵箱和手機立即綁定
已解決430363個問題,去搜搜看,總會有你想問的

使用自定義時間戳提取器的 Kafka 流窗口

使用自定義時間戳提取器的 Kafka 流窗口

精慕HU 2021-11-11 16:42:32
我正在嘗試創(chuàng)建一個 Kafka Streams 應(yīng)用程序,我試圖在一個時間窗口內(nèi)計算每個平臺的唯一設(shè)備。事件類public class Event {    private String eventId;    private String deviceId;    private String platform;    private ZonedDateTime createdAt;}我需要時間窗口尊重事件的 createdAt 所以我寫了一個TimestampExtractor如下的實現(xiàn):public class EventTimestampExtractor implements TimestampExtractor {    @Override    public long extract(final ConsumerRecord<Object, Object> record, final long previousTimestamp) {        final Event event = (Event) record.value();        final ZonedDateTime eventCreationTime = event.getCreatedAt();        final long timestamp = eventCreationTime.toEpochSecond();        log.trace("Event ({}) yielded timestamp: {}", event.getEventId(), timestamp);        return timestamp;    }}最后,這是我的流媒體應(yīng)用代碼:final KStream<String, Event> eventStream = builder.stream("events_ingestion");eventStream    .selectKey((key, event) -> {        final String platform = event.getPlatform();        final String deviceId = event.getDeviceId());        return String.join("::", platform, deviceId);    })    .groupByKey()    .windowedBy(TimeWindows.of(TimeUnit.MINUTES.toMillis(15)))    .count(Materialized.as(COUNT_STORE));當(dāng)我將事件推送到event_ingestion主題時,我可以看到時間戳已記錄到應(yīng)用程序日志中,并且數(shù)據(jù)正在寫入計數(shù)存儲中。當(dāng)我遍歷計數(shù)存儲時,我看到以下內(nèi)容:Key: [ANDROID::1@1539000000/1539900000], Value: 2雖然我的時間窗口是 15 分鐘,但密鑰跨越 10 天。如果我從流配置中刪除我的 TimestampExtractor 實現(xiàn)(因此回到處理時間),密鑰按預(yù)期跨越 15 分鐘:Key: [ANDROID::1@1539256500000/1539257400000], Value: 1我在這里做錯了什么?有任何想法嗎?
查看完整描述

1 回答

?
慕妹3242003

TA貢獻1824條經(jīng)驗 獲得超6個贊

TimestampExtractor 使用紀(jì)元毫秒值進行窗口化。您正在計算“秒”,這會將消息放入錯誤的時間窗口。


查看完整回答
反對 回復(fù) 2021-11-11
  • 1 回答
  • 0 關(guān)注
  • 277 瀏覽
慕課專欄
更多

添加回答

舉報

0/150
提交
取消
微信客服

購課補貼
聯(lián)系客服咨詢優(yōu)惠詳情

幫助反饋 APP下載

慕課網(wǎng)APP
您的移動學(xué)習(xí)伙伴

公眾號

掃描二維碼
關(guān)注慕課網(wǎng)微信公眾號