問題タブ [apache-kafka]

Kafka から読み書きされるメッセージに Avro を使用しようとしています。Avro バイナリ エンコーダーを使用して、メッセージ キューに入れられるデータをエンコード/デコードする例はありますか?

Kafkaの部分よりもAvroの部分が必要です。または、おそらく別のソリューションを検討する必要がありますか？基本的に、スペースに関して JSON のより効率的なソリューションを見つけようとしています。Avro は、JSON よりもコンパクトになる可能性があるため、言及されました。

PythonまたはR（できればR）でkafkaコンシューマーを一緒にハックしようとしています。kafkaコンソールコンシューマーを使用すると、文字列をgrepして関連データを取得できますが、Rで適切に解析することになると途方に暮れます。

他の言語（たとえば、PHP、CPP）で利用可能なkafkaクライアントがありますが、Rの1つは、データ分析の観点から役立ちます。

このフォーラムのエキスパートR開発者が、私がこの方向に進むことを可能にするリソースを示唆/提案できれば素晴らしいと思います。

Apache Kafka：incubator.apache.org/kafka/

Kafkaコンシューマークライアント：https ：//github.com/kafka-dev/kafka/tree/master/clients

Apache-Kafka メッセージング キューについて。

Kafka ダウンロード ページから Apache Kafka をダウンロードしました。に抽出しました/opt/apache/installed/kafka-0.7.0-incubating-src。

クイックスタート ページには、zookeeper を起動してから、次のコマンドを実行して Kafka を起動する必要があると記載されています。
>bin/kafka-server-start.sh config/server.properties

別の Zookeeper サーバーを使用しているため、config/server.propertiesその Zookeeper インスタンスを指すように編集しました。

クイックスタート ページの指示に従って Kafka を実行すると、次のエラーが表示されます。

telnet を使用して、Kafka が実行されているマシンから Zookeeper インスタンスにアクセスできることを確認しました。全て大丈夫。

なぜこのエラーが発生するのですか?

カフカで遊び始めました。Zookeeper 構成を設定し、文字列メッセージを送信して消費することができました。今、私は（Javaで）オブジェクトを渡そうとしていますが、何らかの理由で、コンシューマーでメッセージを解析するときにヘッダーの問題があります。いくつかのシリアル化オプション (デコーダー/エンコーダーを使用) を試しましたが、すべてが同じヘッダーの問題を返します。

これが私のコードですプロデューサー：

そして消費者：

と私のシリアライザ:

そして、これは私が得るエラーです:

MessagePack私が aとプレーンな書き込みで作業したときObjectOutputStream、同様のヘッダーの問題が発生しました。また、メッセージにペイロード CRC32 を追加しようとしましたが、それも役に立ちませんでした。

ここで何が間違っていますか？

基本的な質問ですが、Apache Kafka は現在どのプラットフォームと言語をサポートしていますか?

1日に数ギガバイトのメッセージを処理する必要がある本番環境のアプリケーションがあります。私はKafkaのアーキテクチャとパフォーマンスがとても好きです。それは私のニーズに完全に適合します。

いつかメッセージングレイヤーをKafkaに置き換えたいと思います。0.7.1バージョンは、パフォーマンスの安定性と一貫性の観点から、本番環境での使用に十分ですか？

一度にトピックから 10 件のメッセージを取得するために queuedchunks.max プロパティを使用していますが、利用可能なすべてのメッセージ (10 件以上) を返しています。必要なメッセージのみを取得する方法があれば教えてください。

私はMavenの初心者で、strom kafka TestTopologyスパウトを実行しようとしていますが、ストームカフカスパウトの下のストームカフカcontribでpomファイルが提供されていません。Leinを使用しましたが、どちらも機能しませんでした。私はまた、lein pomを使用して自分でpomを作成しようとしました（そしてそれを変更しました）が、どちらも機能しませんでした. 私はプロジェクトをトップダウンで構築しています: 私は storm-conrib-master で mvn を実行し、それはより高いレベルを正常に構築しますが、どのようにしたら、Storm starter の m2-pom を使用して TestTopology.java を実行できますか?これ：

project.clj設定とleinも使用しましたが、どちらも機能しませんでしたが、leinは私のものではありません。Maven の設定は次のとおりです。

Apache Maven 3.0.4 Maven ホーム: /usr/share/maven Java バージョン:

1.6.0_33、ベンダー: Sun Microsystems Inc.

エンコーディング: UTF-8 OS 名: "linux"、バージョン: "3.2.0-32-generic"、アーキテクチャ: "amd64"、ファミリ: "unix"

TestTopology.java はどのように実行すればよいですか?

コンパイル エラーが発生します。

[エラー] プロジェクト storm-kafka で目標 org.apache.maven.plugins:maven-compiler-plugin:2.0.2:compile (default-compile) を実行できませんでした: コンパイルの失敗: コンパイルの失敗: [エラー] Desktop/storm- contrib-master/storm-kafka/src/jvm/storm/kafka/trident/StaticBrokerReader.java:[13,7] ジェネリックは -source 1.3 ではサポートされていません [エラー] (ジェネリックを有効にするには -source 5 以上を使用してください) [エラー] マップ ブローカー = new HashMap(); [エラー]

現在、このライブラリを使用して、セットアップしたカフカサーバーのストレステストを行っています: https://github.com/dsully/pykafka

結局何が起こっているのかというと、自分のローカル マシンが過負荷になってしまうということです。

エラー 10055が表示されます。これは、Google によると、「一度に開いている接続が多すぎるため、Windows が TCP/IP ソケット バッファを使い果たしました」という意味です。 netstat によると、producer.disconnect() はソケットを閉じているのではなく、TIME_WAIT状態にしています。

ipython デバッガーは次の行を指します。

犯人として、しかし、これは私が快適であるよりも低いレベルで物事を台無しにするようです.

私は検索して、このPythonソケットが接続を適切に閉じていないことを発見しました。

そのため、io.py ファイルでそのオプションを使用して pykafka lib を再構築しました。

それでも同じエラーが発生します。

setsockopt 行を正しい場所に配置していませんか? 他に試してみることはありますか？

コマンド「geminstallfluent-plugin-kafka」により、fluentd用のoutput_kafkaプラグインをインストールしました。

しかし、fluentdサービスを開始すると、ログファイルに次のエラーメッセージが表示されます。

2012-11-09 18:18:39 +0800：一時的にバッファのフラッシュに失敗しました。次の再試行は、2012-11-0918:52:46+0800になります。error="初期化されていない定数Kafka::Message " instance = 69952455476860

output_kafka.rbがモジュールKafkaまたはクラスMessageを見つけられないようですが、どうすれば修正できますか？