問題タブ [apache-kafka]

見つける方法はありますか

• メッセージがまだ消費されていない場所からのオフセットまたはタイムスタンプ

• 消費された最後のタイムスタンプ

または、ストームを使用してカフカで既読メッセージと未読メッセージを区別する

メッセージを消費するためにkafkaスパウトを使用しています。しかし、トポロジを変更してアップロードする必要がある場合、古いメッセージから再開するか、新しいメッセージから開始しますか? Kafka スパウトは、消費する場所からのタイムスタンプを指定しますが、タイムスタンプを知るにはどうすればよいでしょうか?

私はkafka stromの統合に取り組んでいます。エラーで立ち往生しています。を使用して実行しようとすると、ビルドが失敗します
mvn -e -f m2-pom.xml compile exec:java -Dexec.classpathScope=compile -Dexec.mainClass=storm.starter.MainTopology

これは、pom.xml ファイルのスニペットです。

私は試した

私は storm-0.9.0-rc3 と kafka-0.7.2 を使用しています

シナリオ: プロデューサーからコンシューマーへの伝搬遅延を小さくしたい低ボリュームのトピック (~150 メッセージ/秒) があります。

プロデューサーからのタイム スタンプを追加し、それをコンシューマーで読み取って伝播遅延を記録しました。デフォルトの構成では、メッセージ (20 バイト) は 1960 ミリ秒から 1230 ミリ秒の伝播遅延を示しました。同じマシンで 1 つのプロデューサーと 1 つの単純なコンシューマーを試したので、ネットワークの遅延は発生しませんでした。

トピックのフラッシュ間隔を 20 ミリ秒に調整しようとすると、1100 ミリ秒から 980 ミリ秒に低下します。次に、消費者"fetcher.backoff.ms"を10msに調整してみました.1070ms - 860msに落ちました。

問題: 20 バイトのメッセージの場合、伝播遅延をできるだけ低くしたいと考えており、~950 ミリ秒はより高い数字です。

質問: 構成で見逃しているものはありますか? 私はコメントを歓迎します。あなたが得た遅延は最小限です。

仮定: Kafka システムには、コンシューマがプロデューサからメッセージを取得する前にディスク I/O が含まれており、これはハードディスクの RPM などに関連しています。

同じ 20 バイトのメッセージの場合、遅延は 740 ミリ秒から 880 ミリ秒でした。

次のステートメントは、構成自体で明確にされています。
いくつかの重要なトレードオフがあります。

1. 耐久性: フラッシュされていないデータは、クラッシュが発生した場合に失われるリスクが高くなります。
   
2. Latency : データは、フラッシュされるまで消費者に提供されません (これによりレイテンシが追加されます)。
   
3. スループット: 一般に、フラッシュは最もコストのかかる操作です。
   

したがって、150ms ～ 250ms のマークに到達する方法はないと思います。(ハードウェアのアップグレードなし) .

2 つの異なる Linux ボックスで kafka でクラスター環境をセットアップしようとしましたが、実行できませんでした。誰か助けてください。

2 つの Linux ボックスで、デフォルトの Zookeeper.properties ファイルの変更を変更しませんでした。ただし、kafka ルート ディレクトリの config フォルダーの server.properties を変更しました。

Zookeeper.connect でも、zookeeper.connect=server1:2181,server2:2181 として与えようとしましたが、運はありませんでした。

複数のボックスでKafkaを構成する方法を教えてくれる人はいますか、それとも私が間違っているところを教えてくれますか

ありがとう、ビーメッシュ

高レベルのコンシューマー (コンシューマー グループ) を使用して、同じプロセス内の多くのスレッドを使用してトピックを消費する例をたくさん見てきました。複数のプロセス (異なるマシン上) でパーティションを分割し、並行して消費することはできますか? もしそうなら、何か例はありますか？

C で Apache Kafka の PHP 拡張機能を作成しようとしていますが、librdkafkaを使用する必要があります。そこで、kafka クライアント ライブラリをコンパイルしてインストールし、librdkafka/rdkafka.h を含む単純な拡張機能を作成しました。拡張機能は、phpize、./configure、make、make install で問題なくコンパイルできます。しかし、拡張機能を使用しようとすると、エラーが発生します:

つまり、基本的には librdkafka を適切に拡張機能に含めていなかったか、またはこれら 2 つのプログラムのリンクに関連するものを入れていなかったと思います。

ここに私のコードがあります: https://github.com/salebab/phpkafka

librdkafka ファイルは次のとおりです。

何か助けはありますか？

Java Producer API を使用して文字列メッセージを Kafka V. 0.8 に送信します。メッセージのサイズが約 15 MB の場合、MessageSizeTooLargeException. 40 MBに設定しようとしましmessage.max.bytesたが、それでも例外が発生します。小さなメッセージは問題なく機能しました。

(プロデューサーに例外が表示されます。このアプリケーションにはコンシューマーがありません。)

この例外を取り除くにはどうすればよいですか?

私のプロデューサー設定例

エラーログ：

Python ロギング モジュールを多用する大規模で複雑なアプリケーションがあります。

これらのログを Kafka クラスターに取り込み始める必要があり、途中でデータを変更しないようにする必要があります。

私にとって理想的な解決策は、Kafka 用の新しいハンドラーを作成することです。そして、しばらくの間、ログが古いログ ソリューションと kafka の両方に並行して送られるようにします。その後、最終的に古いロギング ハンドラーを停止し、Kafka に送信するだけです。

ただし、kafka ロギング ハンドラーは表示されず、kafka クライアントのみが表示されます。Kafka クライアントを追加するということは、現在のすべてのロギング呼び出しを追跡し、別の呼び出しを新しい kafka クライアントに追加することを意味します。同一の結果を得ることは困難です。

処理するメッセージが少ない場合、Kafka は役に立ちますか。1 秒あたり 1000 件のメッセージを処理する必要がある場合、Kafka は実行可能ですか?