問題タブ [spark-streaming]

Kafka からのメッセージを使用する Spark Streaming ジョブを作成しようとしています。これまでに行ったことは次のとおりです。

1. 飼育係はじめました
2. Kafkaサーバーを開始しました

3. サーバーにいくつかのメッセージを送信しました。以下を実行すると、それらを見ることができます。

   /li>

4. 5 分以内に着信するメッセージの数をカウントするプログラムを作成しようとしています。

コードは次のようになります。

3 番目の引数 (コンシューマー グループ) に使用する値がわからない。これを実行すると、Unable to connect to zookeeper server. しかし、Zookeeper はポートで実行されています2181。そうしないと、ステップ 3 が機能しません。

うまく使えていないようですKafkaUtils.createStream。何か案は？

Kafka と Spark-Streaming の間に問題があります。本番環境で低レベルのトラフィック (1 秒あたり約 12000 ～ 15000 レコード) のサービスを使用しています。最初は、トラフィックの消費は正常に見えますが、10 ～ 15 分後、突然ほぼ1/10残りを消費する速度。ネットワークのトラフィックの問題でしょうか？

Kafka の構成:
num.network.threads=2
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.flush.interval .messages=10000
log.flush.interval.ms=1000
log.retention.hours=12
log.segment.bytes=536870912
log.retention.check.interval.ms=60000
log.cleaner.enable=false
log.cleanup.interval .分=1

スパーク ストリーミングの構成 (コンシューマー):

要素を/に効果的にRDD[T]分割し、元の順序を維持するにはどうすればよいですか?Seq[RDD[T]]Iterable[RDD[T]]n

こんな感じで書けるようになりたいです

次のような結果になるはずです

spark はそのような機能を提供しますか? そうでない場合、これを達成するためのパフォーマンスの高い方法は何ですか?

とても速く見えません..

ワーカー ログを確認したところ、次のエラーが見つかりました。

また、spark.cleaner.ttl の値を 10 分に設定しましたが、うまくいきません。

hdfsフォルダー（sparkstreamining）からストリーミングしようとしています：

それが私がした良いフォルダかどうかを確認するために

JavaDStream にデータが含まれていないという戻り値でエラーが発生していません。フォルダにはすでにいくつかのファイルが含まれています..

フォルダパスを確認する他の方法はありますか? 他にチェックすべき点はありますか？

注：ローカルファイルから読み取ってストリームに入れようとしましたが、まだ空の入力があります（この試行のコードは次のとおりです）。PC上のファイルの場所をspark-shell ..input.print()行がjava.io.Exceptionを与えています...

また、この 2 番目のバージョンのログは、入力に問題があることを示しています。

ありがとうございました！

apache spark ストリーミングで reduceByKey を呼び出すときに、ストリーム内のレコードの順序が保証されているかどうか疑問に思っていました。基本的に、私が行う計算の一部は、最後の値を取得する必要があります。

次に例を示します。