spark-streaming - Receiver が失敗し、WAL が s3 に保存されている場合、Spark ストリーミングは先行書き込みログ レコードからデータを読み取ることができませんでした

Question

ここにエラーログがあります -

org.apache.spark.SparkException: org.apache で先行書き込みログ レコード FileBasedWriteAheadLogSegment(s3n://*****/checkpoint/receivedData/20/log-1439298698600-1439298758600,13678,5069) からデータを読み取ることができませんでした。 spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogBackedBlockRDD.scala:144) at org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD$$anonfun$compute$1.apply (WriteAheadLogBackedBlockRDD.scala:168) org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD$$anonfun$compute$1.apply(WriteAheadLogBackedBlockRDD.scala:168) で scala.Option.getOrElse(Option.scala:120) で org. apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.compute(WriteAheadLogBackedBlockRDD.scala:168) org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:27​​7) org.apache.spark.rdd.RDD.iterator(RDD.scala:244) org.apache.spark.rdd.UnionRDD.compute(UnionRDD. scala:87) org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:27​​7) org.apache.spark.rdd.RDD.iterator(RDD.scala:244) org.apache.spark.rdd .MapPartitionsRDD.compute(MapPartitionsRDD.scala:35) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:27​​7) at org.apache.spark.rdd.RDD.iterator(RDD.scala:244) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35) で org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:27​​7) で org.apache.spark.rdd.RDD.iterator( RDD.scala:244) org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:70) org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41) で org.apache.spark.scheduler.Task.run(Task.scala:70) で org.apache.spark.executor .Executor$TaskRunner.run(Executor.scala:213) で java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) で java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) でjava.lang.Thread.run(Thread.java:745) 原因: org.apache.spark.streaming.util.FileBasedWriteAheadLog.read(FileBasedWriteAheadLog.scala:106) での org.apache.spark での java.lang.NullPointerException。 streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogBackedBlockRDD.scala:141) ... 22 もっと見る70) org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41) で org.apache.spark.scheduler.Task.run(Task.scala:70) で org.apache.spark.executor.Executor $TaskRunner.run(Executor.scala:213) で java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) で java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) で java. lang.Thread.run(Thread.java:745) 原因: org.apache.spark.streaming.util.FileBasedWriteAheadLog.read(FileBasedWriteAheadLog.scala:106) での java.lang.NullPointerException org.apache.spark.streaming. rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogBackedBlockRDD.scala:141) ... 22 もっと見る70) org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41) で org.apache.spark.scheduler.Task.run(Task.scala:70) で org.apache.spark.executor.Executor $TaskRunner.run(Executor.scala:213) で java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) で java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) で java. lang.Thread.run(Thread.java:745) 原因: org.apache.spark.streaming.util.FileBasedWriteAheadLog.read(FileBasedWriteAheadLog.scala:106) での java.lang.NullPointerException org.apache.spark.streaming. rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogBackedBlockRDD.scala:141) ... 22 もっと見る22以上22以上org.apache.spark.executor.Executor$TaskRunner.run(Executor. scala:213) で java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) で java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) で java.lang.Thread.run(Thread .java:745) 原因: org.apache.spark.streaming.util.FileBasedWriteAheadLog.read(FileBasedWriteAheadLog.scala:106) での java.lang.NullPointerException org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache で$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogBackedBlockRDD.scala:141) ... 22 もっと見るorg.apache.spark.executor.Executor$TaskRunner.run(Executor. scala:213) で java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) で java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) で java.lang.Thread.run(Thread .java:745) 原因: org.apache.spark.streaming.util.FileBasedWriteAheadLog.read(FileBasedWriteAheadLog.scala:106) での java.lang.NullPointerException org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache で$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogBackedBlockRDD.scala:141) ... 22 もっと見るrunTask(ShuffleMapTask.scala:41) で org.apache.spark.scheduler.Task.run(Task.scala:70) で org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213) で java .util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) で java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) で java.lang.Thread.run(Thread.java:745) が発生した作成者: java.lang.NullPointerException at org.apache.spark.streaming.util.FileBasedWriteAheadLog.read(FileBasedWriteAheadLog.scala:106) at org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd $WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogBackedBlockRDD.scala:141) ... 22 詳細runTask(ShuffleMapTask.scala:41) で org.apache.spark.scheduler.Task.run(Task.scala:70) で org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213) で java .util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) で java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) で java.lang.Thread.run(Thread.java:745) が発生した作成者: java.lang.NullPointerException at org.apache.spark.streaming.util.FileBasedWriteAheadLog.read(FileBasedWriteAheadLog.scala:106) at org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd $WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogBackedBlockRDD.scala:141) ... 22 詳細scala:70) で org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213) で java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) で java.util.concurrent.ThreadPoolExecutor でjava.lang.Thread.run(Thread.java:745) での $Worker.run(ThreadPoolExecutor.java:615) 原因: org.apache.spark.streaming.util.FileBasedWriteAheadLog.read(FileBasedWriteAheadLog での java.lang.NullPointerException .scala:106) at org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogBackedBlockRDD.scala:141) ... 22 詳細scala:70) で org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213) で java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) で java.util.concurrent.ThreadPoolExecutor でjava.lang.Thread.run(Thread.java:745) での $Worker.run(ThreadPoolExecutor.java:615) 原因: org.apache.spark.streaming.util.FileBasedWriteAheadLog.read(FileBasedWriteAheadLog での java.lang.NullPointerException .scala:106) at org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogBackedBlockRDD.scala:141) ... 22 詳細runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745) 原因: java.lang.NullPointerException org.apache.spark.streaming.util.FileBasedWriteAheadLog.read(FileBasedWriteAheadLog.scala:106) で org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1( WriteAheadLogBackedBlockRDD.scala:141) ... 22 続きを読むrunWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745) 原因: java.lang.NullPointerException org.apache.spark.streaming.util.FileBasedWriteAheadLog.read(FileBasedWriteAheadLog.scala:106) で org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1( WriteAheadLogBackedBlockRDD.scala:141) ... 22 続きを読むscala:106) at org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogBackedBlockRDD.scala:141) ... 22 詳細scala:106) at org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogBackedBlockRDD.scala:141) ... 22 詳細

注: HDFS がストレージとして使用されている場合、WAL からの読み取りは正常に機能します。

どんな助けでも大歓迎です。

Answer 1

現在、s3 では WAL がサポートされていないようで、Spark チームは WAL フリーの構成に取り組んでいます。

https://issues.apache.org/jira/browse/SPARK-9215