問題タブ [apache-kafka]

カフカでトピックをパージする方法はありますか?

大きすぎるメッセージをローカル マシンの kafka メッセージ トピックにプッシュしたところ、エラーが発生しました。

ここで を大きくするのfetch.sizeは理想的ではありません。実際にはそれほど大きなメッセージを受け入れたくないからです。

シンプルな Producer-->Kafka-->Consumer アプリケーションを Java で実装しようとしています。メッセージを正常に生成および消費できますが、コンシューマーを再起動すると問題が発生し、すでに消費されたメッセージの一部が Kafka からコンシューマーによって再び取得されます (すべてのメッセージではなく、最後のいくつかのメッセージ消費されたメッセージ)。

コンシューマを設定autooffset.reset=largestし、autocommit.interval.msプロパティを 1000 ミリ秒に設定しました。

この「既に消費されたメッセージの再配信」は既知の問題ですか、それとも他に不足している設定はありますか?

基本的に、以前に消費されたメッセージが消費者によって取得/消費されないようにする方法はありますか?

次のセットアップのapache-kafka 0.8クラスターがあります-

1) 3 つのブローカーがすべて同じマシンで実行されている
2) 10 個のパーティションと 3 つのレプリカを持つ 1 つのトピック。

1 つのトピックに対して 20 人のプロデューサーがいます。
各パーティションから 10 人の消費者が消費しています。ブローカーのフェールセーフをテストしています。

すべてのブローカーが稼働している場合、消費されるメッセージの数は、生成されるメッセージの数と同じです。

しかし、ブローカーを 1 つずつダウンさせてセットアップをテストすると、生成されるよりも多くのメッセージが消費されることがわかりました。

同じ理由として何が考えられますか?

API を使用して IDE から Kafka でトピックを作成するにはどうすればよいですか。

エラーが発生します：

そして、開発者のセットアップをそのまま踏襲しました。

私はkafka 0.8を試してみたいです（私はそれがすでにリリースされていることを理解しています）。しかし、kafka maven リポジトリはどこにありますか。

また、どのリポジトリ URL を追加する必要がありますか?

私はいくつかのブログを見つけました

しかし、それは機能しません。適切な Maven 依存関係を探しています。それとも、git からチェックアウトして、社内アーティファクトにデプロイする必要がありますか?

状況: Kafka プロジェクトを適切に取得し、Linux コマンド ラインで「クイック スタート」を実行しました。問題はありません。次に、Scala プロジェクトをビルドし (Kafka は Scala にあります)、ソース コードを /core ディレクトリにインポートし、パッケージをビルド パスに配置しましたが、エラーが発生したため、パッケージを main/ にインポートするだけにしました。コアのディレクトリ:

インポート後の状況は次のとおりです。プロジェクトにscalaファイルとjavaファイルがあり、エラーログには、プロジェクトに実際にインポートされたjavaファイルといくつかのscalaファイルが見つからなかったことが示されています。プロジェクトはどうすればいいですか？scala java混合プロジェクトを正しくインポートするには?

新しいサービス (RDBMS 以外のデータ ストアやメッセージ キューなど) を使用するときに最初に考えることの 1 つは、「データをどのように構築すればよいか?」ということです。

いくつかの紹介資料を読み、見ました。特に、たとえばKafka: a Distributed Messaging System for Log Processingを取り上げると、次のように書かれています。

• 「トピックは、メッセージが関連付けられるコンテナです」
• 「並列処理の最小単位はトピックのパーティションです。これは、トピックの特定のパーティションに属するすべてのメッセージがコンシューマー グループ内のコンシューマーによって消費されることを意味します。」

これを知った上で、トピックとパーティションの使用方法を示す良い例は何でしょうか? 何かをいつトピックにする必要がありますか? 何かをいつパーティションにする必要がありますか?

例として、私の (Clojure) データが次のようになっているとします。

トピックはに基づいている必要がありuser-idますか? viewed? at? パーティションはどうですか？

どうやって決めるの？

Flume でできるように、Kafka でファンアウト (複製) できるかどうかは不明です。

Kafka で HDFS または S3 にデータを保存し、リアルタイム処理のためにそのデータの複製を Storm に送信したいと考えています。Storm の集計/分析の出力は、Cassandra に保存されます。Kafka から Storm にすべてのデータを流し、次に Storm から 2 つの出力を流す実装がいくつか見られます。ただし、未加工のデータ ストレージに対する Storm の依存関係を排除したいと考えています。

これは可能ですか？このようなドキュメント/例/実装を知っていますか?

また、Kafka は S3 ストレージを適切にサポートしていますか?

HDFS に保存するための Camus を見ました。このジョブを cron 経由で実行して、Kafka から HDFS にデータを継続的にロードしますか? 前のインスタンスが終了する前にジョブの 2 番目のインスタンスが開始された場合はどうなりますか? 最後に、Camus は S3 で動作しますか?

ありがとうございます。

私は、kafka 経由で hdfs に書き込む必要があるプロジェクトに取り組んでいます。メッセージをkafkaに書き込むオンラインサーバーがあるとします。各メッセージにはタイムスタンプが含まれています。出力がメッセージのタイムスタンプに従ってファイル/ファイルになるジョブを作成したい。たとえば、kafka のデータが

3つのファイルを出力として取得したい

そしてもちろん、このジョブをもう一度実行していて、次のような新しいメッセージがキューにある場合

ファイルを作成する必要があります

私はいくつかのオープン ソースを見てきましたが、それらのほとんどは kafka からいくつかの hdfs フォルダーを読み取ります。この問題に対する最善の解決策/設計/オープンソースは何ですか

問題

各 HA ペアに NFS を使用した複数のデータセンターの activeMQ セットアップがあり、activeMQ は実際にはスケーラブルではなく、NFS の問題でうまく機能しないようです。(5.7を使用しています)

可能な解決策

カフカに移動

要件

• PointToPoint と pub/sub 機能が必要です
• メッセージの優先順位 (kafka がそれをすぐに提供しないことは知っていますが、私たちの側で回避策があります)

質問

これは Kafka で可能ですか (必ずしもすぐに使用できるわけではありませんが、クライアントの調整が必要です)。そうでない場合、他にどのようなテクノロジを提案しますか? JMS である必要はありませんが、スケーラブルで信頼できる必要があります (また、NFS とうまく連携する必要があります)。