apache-kafka

0熱

1答えて

単一のkafkaコマンドを使用して、各トピックのメッセージの合計数を含むすべてのグループのすべてのトピックを一覧表示するにはどうすればよいですか？

すべてのグループIDのすべてのトピックを、各トピックのメッセージの総数とともに一覧表示しようとしていますが、そのようなコマンドはどこにも見つかりませんでした。以下の2つのコマンドを試してみましたが、以下のコマンドの結果を1つのコマンドで期待しています。助けてください。コマンド1：（カフカサーバー内のすべてのトピックを一覧表示する） bin/kafka-topics.sh --list --zoo

1熱

1答えて

アプリケーションの状態ストアをクリアする

カフカストリームアプリケーション内の特定のインスタンスの状態ストアをクリアしたいとします。たとえば、リストのトップ5の最大値を1つの状態に維持しているとします。それを行う方法はありますか？アプリケーションを停止することなく？

0熱

2答えて

Kafka Producerを作成し、send（）、flush（）、close（）メソッドを呼び出す正しい順序は何ですか？

私はカフカのプロデューサーを作り、さまざまな方法を呼び出すという異なるモデルを持っていますが、フローを破らずパフォーマンスに影響を与えないようにプログラムする正しいアプローチが何であるかはわかりません。親切に助けてください。モデル1： for(int i=1; i < 100; i++){ Producer<String, String> producer = new KafkaPro

1熱

1答えて

カフカから最新の値を取得

私はAというカフカのトピックを持っています。トピック内のデータの形式は次のとおりです。今 { id : 1, name:stackoverflow, created_at:2017-09-28 22:30:00.000} { id : 2, name:confluent, created_at:2017-09-28 22:00:00.000} { id : 3, name:kafka, cr

2熱

2答えて

レスポンスカフカで失敗したメッセージを再実行する方法0.8.2.2

処理に失敗した回数x回のメッセージを再発行します。私は手動コミットコードを参照してください：しかし、processMessage（）メソッドで例外を処理する方法はありますか？私は例外を処理し、カフカにメッセージを3回実行するように依頼したいと思います。それでも3回失敗したら、それを捨ててください。

1熱

1答えて

カフカのパーティションとオフセットが消えた

私のKafkaクライアントは、自動スケールが有効になっているGCP App Engine Flex環境で実行されています（GCPはインスタンス数を少なくとも2に保ち、CPU使用率が低いためほとんど2でした）。その2つの仮想マシンで動作するコンシューマ・グループは、数か月にわたって20のパーティションのさまざまなトピックからメッセージを消費していましたが、最近、古いトピックのパーティションは1に縮小

0熱

1答えて

KafkaConsumerとKafkaProducerのアイドル時間設定

私たちのアプリケーションは長時間実行されており、停止/強制終了するまで実行する必要があります。そのための設定は、longのプロデューサとコンシューマの両方にconnections.max.idle.msを設定することです。無限に設定したり、閉じたりしないように設定する方法はありますか？ connections.max.idle.msを10分に設定すると、プロデューサは何も生成されないため、20分間

1熱

3答えて

hbaseの行のデータセットを保存する

私はKafkaからデータ（json文字列）を読み込み、それを解析して与えられたスキーマを持つデータセットに変換し、そのデータセットの属性に対する集計を実行します。私はこのデータセットをjson文字列に変換してからhbaseに格納したいと思っています。これを行う最善の方法は何ですか？サンプルデータセット：HBaseの中に私のキーに対する id|name|age 1 |geet|21 予想さ

3熱

1答えて

kafka - pythonプログラムはhadoopクラスタの外でKafkaに接続できますか？

私はカフカを学んでいて、助言が必要です。私は、ファイルを読み込んで、そのデータをプロデューサーとしてKafkaに書き込むPythonプログラムを作成したいと考えています。 KafkaはHadoopクラスタにセットアップされ、PythonはHadoopクラスタの外で実行されます。外部から接続することはできますか？私はPythonのKafkaライブラリをインストールする必要がありますか？どのよう

1熱

1答えて

KafkaStream createTopicはKafkaサーバーのauto.create.topics.enable設定を尊重していません

私たちは最近、新しいトピックの束で汚染された生産カフカクラスターを持っています。カフカのクラスタは、次の設定があります。調査時 auto.create.topics.enable=false delete.topic.enable=false 、私はこれらのトピックはKafkaStreamのcreateTopicメソッドを使用しているクライアントのチームによって作成されたことがわかった。 o