問題タブ [spark-streaming]

Spark ストリーミングのドキュメントには、次のように記載されています。

Spark Streaming アプリケーションには、受信したデータを処理し、レシーバーを実行するのに十分なコアを割り当てる必要があることを覚えておくことが重要です。

その後：

アプリケーションに割り当てられたコアの数が入力 DStreams/レシーバーの数以下の場合、システムはデータを受信しますが、それらを処理することはできません。

なんらかの方法で阻止されない限り、CPU コアの数に関係なく、アプリケーションが進行するように OS が CPU をスケジュールするため、これは驚くべきことです。私の質問は次のとおりです。

• Spark は通常の CPU スケジューリングを防ぐために何か特別なことをしますか?
• もしそうなら、その背後にある合理性は何ですか？

Spark SQL と Spark Streaming を統合するときの Not Serializable 例外に加えて

私のソースコード

JavaSQLContext も ForeachRDD ループの外で宣言されていますが、まだ NonSerializableException が発生しています

14/12/23 23:49:38 エラー JobScheduler: ジョブ ストリーミング ジョブの実行中にエラーが発生しました 1419378578000 ms.1 org.apache.spark.SparkException: org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala でタスクをシリアル化できません:166) org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158) で org.apache.spark.SparkContext.clean(SparkContext.scala:1435) で org.apache.spark.rdd.RDD .map(RDD.scala:27​​1) at org.apache.spark.api.java.JavaRDDLike$class.map(JavaRDDLike.scala:78) at org.apache.spark.sql.api.java.JavaSchemaRDD.map(JavaSchemaRDD) .scala:42) で com.basic.spark.NumberCount$2.call(NumberCount.java:79) で com.basic.spark.NumberCount$2.call(NumberCount.java:67) で org.apache.spark.streaming. api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:27​​4) at org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:27​​4) at org.apache.spark.streaming.dstream.DStream$ $anonfun$foreachRDD$1.apply(DStream.scala:529) at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1.apply(DStream.scala:529) at org.apache.spark.streaming. dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:42) at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40) at org.apache. spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40) で scala.util.Try$.apply(Try.scala:161) org.apache.spark.streaming.scheduler.Job.run で(Job.scala:32) org.apache.spark.streaming.scheduler で。JobScheduler$JobHandler.run(JobScheduler.scala:171) で java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) で java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) で java .lang.Thread.run(Thread.java:724) 原因: java.io.NotSerializableException: org.apache.spark.sql.api.java.JavaSQLContext at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1181) java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1541) で java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506) で java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429) で java.io. java.io.ObjectOutputStream の ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175)。defaultWriteFields(ObjectOutputStream.java:1541) で java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506) で java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429) で java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java) :1175) で java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1541) で java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506) で java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429) で Java .io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175) で java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347) org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:42) で org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:73) at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:164) ... 20 以上

何か提案があればよろしくお願いします。

kafka コンシューマの実行中に次のエラーが発生しました:

build.sbt ファイル:

エラーの原因は何ですか?

このトライデントの仕事があるとします：

Spark Streaming で同じことを達成するにはどうすればよいですか? 私は見ましupdateStateByKeyたが、これは状態を内部的に保持しているように見えました（Memcachedのような外部状態に保持するのではなく）＆無期限に。saveAsTextFileまた、そのバッチで更新されたキー値のみを発行するのではなく 、たとえば保存すると、各バッチですべてをダンプしようとするようです。

を使用して外部状態と簡単にやり取りできることはわかってforeachRDDいますが、その場合、レコードを正確に 1 回処理するにはどうすればよいでしょうか?

ログを継続的に生成する Tableau サーバー (Windows プラットフォーム) があります。リアルタイム分析のためにApache Sparkにストリーミングしたいと思います。次の解決策を見ましたが、要件を満たすものはないようです。

1) nxlog エージェントを使用します。将来、複数のタブロー サーバーからログが取得される可能性があるため、これはスケーラブルではありません。

2) 流暢、flume は Windows と互換性がありません。

3) Kafka は、ログ ファイルを追跡しないため、論外です。

そのような問題に対するスケーラブルな解決策は何ですか? 主な制限は、Tableau Server が Windows で実行されていることです。

Scala で Spark Streaming を使用しようとしていますが、エラーが発生し、その理由がわかりません。
StreamingContext は、エラーを示す行です。

これらは 2 つのエラーです:

シンボリック参照が正しくありません。StreamingContext.class の署名は、利用できないパッケージ org.apache.hadoop の用語 conf を参照しています。現在のクラスパスから完全に欠落しているか、クラスパスのバージョンが StreamingContext.class のコンパイル時に使用されたバージョンと互換性がない可能性があります。

と：

クラス ファイル 'StreamingContext.class' の読み込み中に、欠落または無効な依存関係が検出されました。パッケージ org.apache.hadoop の term conf にアクセスできませんでした。それ (またはその依存関係) が見つからないためです。依存関係の欠落または競合がないか、ビルド定義を確認してください。(-Ylog-classpath を指定して再実行し、問題のあるクラスパスを確認してください。) 「StreamingContext.class」が互換性のないバージョンの org.apache.hadoop に対してコンパイルされた場合は、完全な再構築が役立つ場合があります。

この質問は以前に尋ねられました: Spark Streaming StreamingContext エラー エラーは依存関係の問題から来ているようですが、私の依存関係がすべて順調であることがわかる限り.