apache-kafka-streams

0熱度

1回答

KeyValueStore<String, Long> kvStore=(KeyValueStore<String, Long>) Stores.create("InterWindowStore1").withKeys(Serdes.String()) .withValues(Serdes.Long()) .persistent() .build(

1熱度

2回答

卡夫卡流：一個記錄到多個記錄

鑑於：我在卡夫卡有兩個主題讓我們說主題A和主題B.卡夫卡流從主題A讀取記錄，處理它併產生多個記錄（比如說recordA和recordB）對應於消耗的記錄。現在，問題是如何使用Kafka Streams來實現這一點。 KStream<String, List<Message>> producerStreams[] = recordStream.mapValues(new ValueMapper<Me

2熱度

1回答

卡夫卡流和卡夫卡表一對多關係加入

我有一個kafka流 - 說博客和卡夫卡表 - 說與這些博客相關的評論。卡夫卡流的關鍵字可以映射到卡夫卡表中的多個值，即一個博客可以有多個評論。我想要對這兩者進行連接，並用一系列註釋ID創建一個新對象。但是當我進行連接時，流只包含最後一個註釋ID。是否有任何文檔或示例代碼可以指出我如何實現這一目標？基本上，是否有任何文檔闡述如何使用Kafka流和Kafka表進行一對多關係連接？ KStream<I

0熱度

1回答

如何使用KeyValueMapper將輸入KStream <String，String>映射到<String，CarClass>？

接收我只想在CarClass上映射的Json數據，並想創建新的流，但映射方法不允許我映射自定義數據類型類型KStream中的方法映射（KeyValueMapper>）不適用於參數（新的KeyValueMapper>（）{}）？

1熱度

1回答

在卡夫卡流，這是什麼VALUE_SERDE_CLASS_CONFIG做

在卡夫卡流，什麼是目的： props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); ...考慮到SE

1熱度

1回答

卡夫卡流給怪異輸出

我玩卡夫卡流試圖做基本聚合（爲這個問題的目的，每個消息只增加1）。在接收完成的KTable變化的輸出話題，我變得很怪異輸出： @B� @C @C� @D @D� @E @E� @F @F� 我認識到，「」表示，它的打印出某種性格並不在字符存在設置，但我不知道爲什麼。這裏是我的代碼，以供參考： public class KafkaMetricsAggregator {

0熱度

1回答

加入兩種不同KStreams

值所以，我有兩個不同的KStream就像這樣：流1：（String鍵，Object值1）流2：（String鍵，Object值2）我想加入他們，這樣我就會看到一個類似於（Object value1，Object value2）的流。乾淨的方法是什麼？

1熱度

2回答

Gui查看器的RocksDb sst文件

我正在與Kafka一起將數據保存到rocksdb中。現在我想看看由Kafka創建的db鍵和值。我下載了FastNoSQL並嘗試失敗。該文件夾包含：文件名爲.sst .log文件當前文件標識文件 LOCK文件 LOG文件清單文件選項文件我該如何看值？

3熱度

1回答

Kafka Stream：首先輸出到主題或直接持續？

有相當數量的文章描述瞭如何使用Kafka Streams來輸出新的Kafka主題，而不是保存到某種分佈式數據庫。這只是一個常見用例，假定嵌入式db +交互式查詢已經足夠，或者有一些體系結構的原因，爲什麼在再次使用主題之前需要輸出主題才能保留它，而不是持久化直？我不確定它是否有所作爲，但我所看到的示例的上下文是用於翻轉時間窗聚合。

2熱度

2回答

從一個主題中的特定分區（Kafka Streams）流式傳輸

據我瞭解，在閱讀Kafka Streams文檔後，無法將其用於從給定主題中的一個分區流式傳輸數據，因此總是需要將其全部讀取。這是正確的嗎？如果是這樣，是否有計劃在將來爲API提供這樣的選項？