kafka-consumer-api

0熱

2答えて

Spring Kafkaパーティション用の複数のコンシューマ

私たちは、スプリングカフカを使用してメッセージを消費しています。メッセージを受け取るために各パーティションの受信機を作成しました。今では、単一のパーティションからコンシューマメッセージに複数のレシーバが必要な要件があります。 パーティション0があるとします。現在、このパーティションからメッセージを受信するレシーバ（Receiver 1）は1つだけです。 これで、同じパーティション（パーティション0

apache-kafka kafka-consumer-api spring-kafka 2017-09-10

---

0熱

2答えて

Apache Kafkaで追加されたパーティションを消費する

私はKafkaConsumerを持っており、手動でパーティションを割り当てています。パーティションを配布するには、追加のパーティションを検出するために一定の間隔でconsumer.partitionsFor(topicId)を使用します。ジョブは永遠に実行され、このケースをサポートしたいからです。ただし、コンシューマを再起動しない限り、これは常にパーティションの初期リストを返します。 消費者から追

apache-kafka kafka-consumer-api 2017-09-12

---

0熱

2答えて

Kafka Brokerと同じマシンでKafka Streamsアプリケーションを実行できますか？

私は、いくつかのトピックからデータを取り込み、データを結合して別のトピックに配置するKafka Streamsアプリケーションを使用しています。 カフカ構成： 5 kafka brokers Kafka Topics - 15 partitions and 3 replication factor. 注：私は私のカフカブローカーが実行されている同じマシン上でカフカストリームアプリケーションを

apache-kafka kafka-consumer-api kafka-producer-api apache-kafka-streams 2017-09-12

---

0熱

1答えて

Kafkaストリームスタートアップの問題 - org.apache.kafka.streams.errors.LockException

私はKafka Streamsアプリケーションのバージョン - 0.11を使用して、いくつかのトピックからデータを取り出し、データを結合して別のトピックに配置します。 カフカ構成： 5 kafka brokers - version 0.11 Kafka Topics - 15 partitions and 3 replication factor. レコードのほとんど何百万人が消費されてい

apache-kafka kafka-consumer-api apache-kafka-streams 2017-09-13

---

0熱

1答えて

Flumeを使用したKafkaメッセージで予期しない文字が発生しました

Flume & Kafkaを使用してCSVファイルを消費し、FlumeインターセプタでJSONのイベントを変換し、Kafkaでプッシュすることで、 私はKafkaに送信される前にメッセージを記録していますが、通常の有効なJSONです。しかし、カフカから同じメッセージを消費すると、シリアル化しようとするとエラーが発生し、有効なJSONではないと言っています。 例えば： は確かに私は私のメッセージの先

apache-kafka cloudera kafka-consumer-api flume flume-ng 2017-09-14

---

2熱

1答えて

kafkaコンシューマグループのトピックを更新する方法を効率的に実装する方法

複数のカフカトピックの情報を処理する「警告」アプリケーションを作成しようとしています。何千ものトピックが存在しますが、アラート設定に基づいて、現時点では何百もの処理が必要です。トピックリストを「定期購読」で継続的に更新すると、リバランスの遅延によってアラートが遅れることがあります。 絶え間なく変化するトピックに登録している消費者グループを効率的に実装するにはどうすればよいですか？

performance apache-kafka alert kafka-consumer-api 2017-09-15

---

0熱

1答えて

RD Kafka Clientのファイルから設定値を読み取る方法は？

私はLIBRDKAFKA Cライブラリを使用して1つのRDカフカクライアントを開発しました。 LIBRDKAFKA Cライブラリには、consumerおよびProducer Handlesの設定値を設定するためのset_config APIが公開されています。設定できる利用可能な設定は数百もあり、それぞれの値に対してset_congigが非常に面倒になりました。重要なのは、ここではコード自体にco

kafka-consumer-api kafka-producer-api 2017-09-15

---

0熱

1答えて

カフカ+スパークストリーミング：kafka.common.OffsetOutOfRangeException

私はこの全体のカフカ/スパーク事に新たなんです。私はSpark Streaming（PySpark）がKafkaプロデューサーからデータを取り込んでいます。それは1分で正常に実行され、その後常にkafka.common.OffsetOutOfRangeExceptionをスローします。 Kafkaのコンシューマはバージョン0.8です（PySparkの場合、0.10はサポートされていません）。 AW

apache-spark pyspark apache-kafka spark-streaming kafka-consumer-api 2017-09-15

---

0熱

1答えて

カフカに関するすべてのメッセージを消費して切断します。

1日に1回トリガーされるバッチジョブがあります。 が プロセスのメッセージ処理が正常に完了した場合 、オフセットをコミット、その時点でカフカトピックで使用可能なすべてのメッセージを消費する必要があります。 現在、ConsumerRecords.isEmpty（）がtrueになるまでwhileループ内のメッセージをポーリングします。 ConsumerRecords.isEmpty（）がtrueの場合

java apache-kafka kafka-consumer-api 2017-09-16

---

0熱

1答えて

Kafkaスパークストリーミングの読み込みと処理の並列処理

私は、Kafkaメッセージの読み込みを並列化して、並列処理しようとしています。私のカフカのトピックには10のパーティションがあります。私は5つのDStreamsを作成し、Unionメソッドを適用して単一のDStreamで操作しようとしています。これまでに試したコードは次のとおりです。 def main(args: scala.Array[String]): Unit = { val p

scala apache-spark apache-kafka spark-streaming kafka-consumer-api 2017-09-16

---