apache-kafka-streams

1熱

1答えて

Kafka Streams：リストア中に変更されたログ状態が変化するため、再調整に失敗しました

Kafkaストリームコンシューマの1人で受信した例外を理解するための助けが必要です。低レベルのプロセッサAPIを使用してKafkaストリームを実装しました。 Kafkaから受け取った更新ごとに、キーストアにマージされ、更新され、状態が維持されます。当初は消費者は1人しかいませんでしたが、しばらくして2番目の消費者を呼び出そうとしました。しかし、再調整の間に2番目の消費者は、再バランスに失敗したと

0熱

1答えて

カフカのカウント数が更新されていない

私はカフカストリームを実験し始めています。私はhttps://kafka.apache.org/0110/documentation/streams/quickstartに従っています。私のサンドボックスは、Ubuntu 16.04.2 LTS、Kafka 0.11.0.0およびScala 2.11.11を実行するボックスです。としては、カフカストリームクイックスタートガイドで説明し、ここで私

1熱

1答えて

StateStoreとは別のアプリケーションからKTableにアクセスできない

ドッカーの単一インスタンス環境で別のアプリケーションからKTableにアクセスする方法をテストする2つのJavaアプリケーション（App1、App2）があります。最初のApp（App1）は、次のコードでKTableに書き込みます。 public static void main(String[] args) { final Properties props = new Pr

1熱

1答えて

kafka-streamsを使用して複数の集約を含む新しいKStreamを作成する

Webサービスのリクエストと応答に関する詳細を含むJSONメッセージをKafkaトピックに送信しています。私はKafkaストリームを使ってKafkaに到着するごとに各メッセージを処理し、その結果をクライアントが接続されているWebSocketに継続的に更新された要約（JSONメッセージ）として送信したいと考えています。クライアントはJSONを解析し、Webページにさまざまなカウント/サマリーを表

1熱

1答えて

何を使用すればいいですか：Kafka StreamまたはKafka consumer APIまたはKafka connect

KafkaストリームまたはKafkaコンシューマAPIまたはKafkaが接続するには、私にとって何が最善かを知りたいですか？トピックからデータを読み取り、処理してデータベースに書きたいと思います。だから私は消費者を書いたが、私はKafkaストリームアプリケーションを書くことができ、それをステートフルなプロセッサーを使って変更してデータベースに書き込むことができ、私の消費者コードを取り除くことがで

1熱

2答えて

カフカは私がScalaの下にでストリームカフカを使用しようとしていますスカラ座

でストリームを完全に正常に動作するJavaでの私のコードされて次のように私のScalaはコードである KStreamBuilder builder = new KStreamBuilder(); KStream<String, String> textLines = builder.stream("TextLinesTopic"); textLines.foreach((key

1熱

1答えて

Apache Kafka：プロセッサは入力ごとに複数の出力を生成します

私は入力として1つのアイテムを取らなければならないカフカストリームプロセッサを持っており、出力として複数のアイテムを生成します。コードを作成する正しい方法は何ですか？ this.context().forward(key, item)への複数回の呼び出しが正しい方法ですか、別の不正行為がありますか？ありがとうございました。