apache-kafka-streams

0熱度

1回答

多流應用我有2個分區和2所相同application.id和不同的接收器主題卡夫卡流應用的輸入話題。 1）我開始兩個應用程序一起，他們開始接收數據，但是當我殺的應用之一，另外一個一直陷入在此。（僅供參考，消費者設置已SESSION_TIMEOUT_MS_CONFIG設置爲120000） [2017年10月8日00：26：17297] DEBUG [卡夫卡協調器心跳線程| ]發送對組心跳請求tes

1熱度

2回答

卡夫卡流中線程分配的策略是什麼？

我的代碼做更多的少這樣的設置： // loop over the inTopicName(s) { KStream<String, String> stringInput = kBuilder.stream(STRING_SERDE, STRING_SERDE, inTopicName); stringInput.filter(streamFilter::passOrFilterMessag

0熱度

1回答

將KTable數據放入外部數據庫

我有一個場景，我需要將數據放入KTable中，並將該數據從KTable存儲到Oracle 11g數據庫中。什麼是可能的方式來做到這一點？

0熱度

1回答

在KafkaStreams中使用SessionWindows聚合數據（0.11）

我想在Kafka（0.11）中的聚合函數中使用SessionWindows，但無法理解，爲什麼我會收到錯誤。這裏是我的代碼片段： // defining some values: public static final Integer SESSION_TIMEOUT_MS = 6000000; public static final String INTOPIC = "input"; pu

0熱度

1回答

Kafka Streams - 刪除/清除狀態存儲

我有一個自定義Transformer實施在我的kafka流DSL的結尾，並持續更新日誌KeyValueStore綁定到它。自從幾周以來，我一直在商店中放置太多數據。現在，無論何時加載應用程序，它都會消耗太多內存。但是，應用程序本身只是一個原型，所以我不介意完全清理商店。我可以重命名kafka.application.id和state-store-name，但這是一個臨時解決方法（並且相應的數

0熱度

1回答

一對多在Kafka中使用KTable的外部聯接

我有一個聯繫人發佈到主題＃1，其中的鍵爲contact.id。此外，還會將一個子表發佈到主題＃2，其中鍵爲subtable.id。現在，我需要做一個連接，如 select * from contact c outer join subtable st on c.id = st.id; 表之間有1..n的關係。如何在此處執行帶/不帶lambda的聚合。

0熱度

2回答

如何從流卡夫卡到卡桑德拉和增量計數器

我有Apache的訪問日誌文件，我想每個頁面的存儲訪問數（總/每天/每小時）在卡桑德拉表。我試圖通過使用連接卡夫卡從日誌文件流的卡夫卡話題做。爲了增加Cassandra中的度量指標，我可以再次使用Kafka Connect嗎？否則，這裏應該使用哪種其他工具，例如kafka流，火花，flink，kafka連接等？

0熱度

2回答

是否有意義使用Apache卡夫卡對於這種情況？

有幾個應用程序必須集成在一起，他們必須交換問題。所以他們中的一個會解決問題，然後做一些事情，然後改變這個問題的狀態。其他可能涉及此問題的應用程序應該獲得新的信息。這一直持續到問題達到最終狀態結束。問題是必須映射問題，因爲這些應用程序並不都支持相同的數據格式。我不知道是否總是發送整個問題還是隻發送新狀態作爲事件。卡夫卡如何支持數據轉換？如果我的問題有一個附件？（> 5MB）謝謝你的建議

3熱度

1回答

Kafka Streams：如何更改記錄時間戳（0.11.0）？

我正在使用FluentD（v.12上一個穩定版本）向Kafka發送消息。但FluentD使用的是舊的KafkaProducer，因此記錄時間戳記始終設置爲-1。因此，我必須使用WallclockTimestampExtractor將記錄的時間戳設置爲時間點，當郵件到達kafka時。時間戳我在真的有興趣，由fluentd消息中發送：「時間戳」：「1507885936」，「主人」：「V.X

2熱度

1回答

是否可以使用Kafka Streams訪問消息標題？

在卡夫卡0.11中添加了Headers到記錄（ProducerRecord & ConsumerRecord），在使用Kafka Streams處理主題時是否可以獲取這些標題？當在KStream調用諸如map方法提供了key和記錄的value的論點，但沒有辦法，我可以看到訪問headers。如果我們可以通過map而不是ConsumerRecord這將是很好的。 ex。 KStreamBuilder