apache-kafka-streams

0熱

1答えて

Kafka Streamsでは、私はJoinWindowを使用してストリーム/ストリーム結合を使用しています。私が理解できるところでは、JoinWindowsはStoresを使用して実装され、その後両方のストリームのデータはRocksDBデータベースに格納されます。しかし、イベント時間が終了すると（で定義された保存期間を（）まで過ぎたら、物理データはどうなりますか？それらはRocksDBデータ

0熱

1答えて

どのようにカフカがコンシューマーオフセットをコミットしましたか？

私はカフカストリームアプリケーションを持っています。私のアプリケーションがイベントを正常に処理しています。イベントを再処理/スキップするために必要なオフセットを使用して、カフカコミット済消費者オフセットを変更する方法。私はHow to change start offset for topic?を試しました。しかし、私は 'Node does not exist：'というエラーが出ます。私を助け

0熱

1答えて

依存関係が見つかりません。バージョンのエラーを示しています

私はkafkaストリームの依存関係をpom.xmlに含めることを試みていますが、私はintelij idea tool ..を使用していますが、これらの依存関係は見つからないとエラーを表示しています。それはkafkaストリーム、slf4j apiとslf4j log4j12の依存関係のエラーを表示しています。誰も問題を助けることができますか？ <?xml version="1.0" encodin

0熱

2答えて

JSON用のCreatin kafkaストリームAPI

JSON配列をJSON要素に変換するためのkafkaストリームコードを作成しようとしています...私はkafkaストリームを初めて使用しています。そこkstreamとktable .. と入力の私のストリームにおける次の形式 [ {"timestamp":"2017-10-24T12:44:09.359126933+05:30","data":0,"unit":""}, {"timestamp

0熱

1答えて

KafkaでflatMapValuesを使用する方法

ScalaでflatMapValuesをKafkaライブラリで使用するとエラーが発生します。ここに私のコードは次のとおりです。 val builder: KStreamBuilder = new KStreamBuilder() val textLines: KStream[String, String] = builder.stream("streams-plaintext-input")

0熱

1答えて

設定ウィンドウ[ホッピング、タンブリングなど]カフカストリームの並列性

kafkaストリームでホッピングウィンドウコードを書いています。minMaxCalculator()は、ストリーム内の最小値と最大値を一度キーでグループ化しています。 KTable<Windowed<String>, aggrTest> WinMinMax = Records.groupByKey().aggregate(new aggrTestInitilizer(), new mi

3熱

2答えて

カフカストリーミングアプリケーションをKafkaクラスタに展開する方法

私はKafkaストリーミングアプリケーションを作成しました。私はKafkaクラスタに展開したいと思います。だから私は、jarファイルを構築し、コマンドを使用して、それを実行します。 java -jar KafkaProcessing-1.0-SNAPSHOT-jar-with-dependencies.jar testTopic kafka1:9092,kafka2:9092 zookeeper1

0熱

1答えて

KStreamの互換性のないタイプ

文字列を使用してコード部分で何かを試していましたが、以下のコードを思いついたので、コードの構文上の問題を助けてください。二KStream文で KStreamBuilder builder = new KStreamBuilder(); KStream<String, String> textlines = builder.stream("INTOPIC"); KStream<String, S

0熱

1答えて

KafkaStreamsアプリケーション

私は別々の2つのkafkaストリームアプリケーションを別々のJVMインスタンスにそれぞれ実行し、うまく動作しています。同じJVMインスタンスでアプリケーションを実行すると、2番目のアプリケーションは動作していません（データの消費もデータの生成もありません）。同じJVMインスタンス内で別々の2つのアプリケーションを実行する場合に制限はありますか？これはカフカの消費者にも起こりますか？