問題タブ [confluent-platform]

コンフルエントなkafka-restシリアライザーを変更することは可能ですか? コンフルエントな avro シリアライザーをカスタマイズしたシリアライザーに交換したいのですが、プロデューサー プール内を確認しました ( https://github.com/confluentinc/kafka-rest/blob/2.x/src/main/Java/io/confluent /kafkarest/ProducerPool.java )。

buildAvroProducer をビルドするときに、コンフルエント ライブラリである KafkaAvroSerializer を要求します。そのため、コードを変更せずにカスタマイズすることはできないようです (クラスパスまたは sth で単に jar を交換するなど)。

を使用してストリーミング データを に書き込みConfluentます。ユーザーマニュアルに従い、コネクタのクイックスタートとセットアップを行いました。トピックを 1 つだけ使用すると、正常に動作します。私のプロパティファイルは次のようになりますHDFS ConnectorHDFS

複数のトピックを追加すると、オフセットが継続的にコミットされ、コミットされたメッセージが書き込まれません。

1と2でtasks.maxを試しました。以下のように連続してCommitting offsetsログに記録されます

サービスを正常に停止すると (Ctrl+C)、tmpファイルが削除されます。私は何を間違っていますか？それを行う適切な方法は何ですか？これに関する提案をお待ちしております。

avro ペイロードを kafka-rest サービスに発行するときに、value_schema のフィールドの 1 つがデフォルト値で定義され、records でこのフィールドが省略されている場合。kafka-rest はまだレコード内に保持することを主張しているようです。そうする正しい方法は何ですか？

例えば。

出力

現在、OnMessage イベント ハンドラーの Simple/Advanced Consumer の例に従っていますが、パフォーマンスはかなり貧弱です。(1 分あたり 120,000 メッセージ vs (C++ ライブラリを使用すると 1 分あたり 100 万メッセージ)

ビジー ループを実装して .Consume(TimeSpan) を使用するのではなく、OnMessage イベントを待機することで、メッセージを消費する際の遅延が発生する可能性がありますか?

0.9 kafka に接続するには、confluent schema-registry を使用する必要があります。しかし、schema-registry 2.0.1 バージョンでは、利用可能な SSL 構成が表示されませんでした。schema-registry と kafka-rest の ssl が 0.9 kafka と通信できるようにする方法はありますか?

カフカ接続にelasticsearchコネクタ(コンフルエント)を使用しています。トピックからelasticsearchインデックスにメッセージを取得しています。キーが使用されていない場合、オフセットは _id の一部であることがわかります。オフセットとタイムスタンプに基づいて、elasticseach のメッセージを閲覧したいと考えています。

メッセージのオフセットとタイムスタンプをインデックスのフィールドとして取得することは可能ですか?

これがelasticsearchドキュメントです

{ "_index": "test-elasticsearch-sink", "_type": "kafka-connect", "_id": "test-elasticsearch-sink+0+0", "_score": 1, "_source": { "f1": "value1" } }

ありがとう、ラジェッシュ

プロセス中に、ハイブ統合をオンにして、kafka-connect を使用して HDFS にデータをストリーミングしようとしました。

私のユースケースでは、「FieldPartioner」をパーティショナー クラスとして使用する必要があります。

私の問題は、複数のパーティションを取得できないことです。

例：

私の例のJSON

「mydate」と「hour」に基づいてパーティションを作成したい

私は次のことを試しました

また、partition.field.nameを次のように指定してみました

と

そして、より多くのそのような組み合わせ

この問題に関するヘルプは大歓迎です

ありがとう。

JSON で Kafka との間で生産/消費します。以下のプロパティを使用して、JSON で HDFS に保存します。

プロデューサー :

消費者 ：

問題-1:

例外を取得:

問題-2:

上記の 2 つのプロパティを有効にしても問題は発生しませんが、hdfs にデータが書き込まれません。

どんな提案でも大歓迎です。

ありがとう