問題タブ [kafka-python]

次のコードで、グループ G1 の 1 つの Kafka コンシューマのフェッチ オフセットを off1 に巻き戻しました。

上記の行で、off1 は、対応するパーティション内のトピックで使用可能な最も古いメッセージのオフセットです。

今、次のように別のグループ G2 で Kafka コンシューマーをインスタンス化しています。

ここでは、フェッチ オフセットを off1 として読み取っています。これは、グループ G1 の consumer1 に対してリセットしたものと同じです。異なるグループのオフセットは異なるはずなので、それは起こらないはずだと思います。誰かが明らかにしたら、私は本当に感謝します. 前もって感謝します。

Python Kafka トピックを使用しています。

Kafka のキュー内のメッセージを更新して、キューの先頭に再度追加できるプロビジョン プロデューサーはありますか?

Kafka の仕様によると、実現可能ではないようです。

Python kafka-library で SimpleProducer を使用しています。このスクリプトは、私が試した他のよりハードに構成された kafka セットアップで以前に問題なく動作しました。

このスクリプトを 1 回実行すると、python-console にこの応答が表示されます。

次に、zookeeper.log のノードに移動して、次を確認できます。

これは、Zookeeper がトピック用の新しい Znode を作成しているだけのようです。これは以前から存在していなかったからです。また、Kafka server.log には次のように出力されます。

ただし、私のメッセージはトピックに投稿されることはなく、次に python-script を実行すると、常に次のようになります。

私がそれを機能させた場合、advertized.host.name は常にノードの外部 IP でしたが、Kubernetes を介してそれを機能させることができないようです。外部 IP をコンテナー parhaps から呼び出し可能にすることは可能でしょうか?

私の kafka/config/server.properties は、すべてのブローカーで次のようになります。

キーに基づいてデータを分離するには、同じトピック内で複数のトピックまたは複数のパーティションを使用する必要がありますか? サーバーで発生するオーバーヘッド、計算、データストレージ、および負荷に基づいて質問しています。

私はカフカが初めてです。csv ファイルから Kafka にデータをインポートしようとしています。前日のデータが廃止されている間、毎日インポートする必要があります。PythonでKafkaトピックの下にあるすべてのメッセージを削除するにはどうすればよいですか? または、PythonでKafkaトピックを削除するにはどうすればよいですか? または、誰かがデータの有効期限が切れるまで待つことを提案しているのを見ました。可能であれば、データの有効期限を設定するにはどうすればよいですか? どんな提案でも大歓迎です！

ありがとう

PubMed データ ソースでは、出力を Kafka キューにプッシュする必要があります。各ソースは Kafka トピックとして表示できます。(私は Kafka の概念を知っており、Python を使用して Kafka を探索しました)

FireFTP を介して PubMed データを表示できます。

誰でも前進する方法を助けることができますか?

このメソッドは Kafka ソースで見つけました。使うべきですか？

kafka イベントを消費するプログラムを作成しました。10秒後に終了したいデーモンがあります。

インデントは無視してください。
しかし、このプログラムは 10 秒後に終了しません。ここで何が欠けているのか知りたいですか?

次のコードは、毎秒 10k ～ 20k のレコードを取り込み、パフォーマンスを改善したいと考えています。json 形式を読み取り、Kafka を使用してデータベースに取り込みます。- Zookeeper と Kafka がインストールされた 5 つのノードのクラスターで実行しています。

改善するためのヒントを教えていただけますか？