执行器失败后 Spark 无法在 HDFS 中找到检查点数据

Question

我正在从 Kafka 发送数据，如下所示：

final JavaPairDStream<String, Row> transformedMessages = 


    rtStream
                    .mapToPair(record -> new Tuple2<String, GenericDataModel>(record.key(), record.value()))                
                    .mapWithState(StateSpec.function(updateDataFunc).numPartitions(32)).stateSnapshots()                        
                    .foreachRDD(rdd -> {
                    --logic goes here
                    }); 

我有四个工作线程和这个应用程序的多个执行程序，我正在尝试检查 Spark 的容错性。

由于我们使用 mapWithState，spark 正在检查点数据到 HDFS，所以如果任何 executor/worker 出现故障，我们应该能够恢复丢失的数据（数据丢失在死 executor 中），并继续使用剩余的 executor/workers .

所以我杀死了一个工作节点以查看应用程序是否仍能顺利运行，但我在 HDFS 中得到了 FileNotFound 异常，如下所示：

这有点奇怪，因为 Spark 有时在 HDFS 中检查点数据，为什么它无法找到它。显然 HDFS 并没有删除任何数据，所以为什么会出现这个异常。

或者我在这里遗漏了什么？

[ERROR] 2018-08-21 13:07:24,067 org.apache.spark.streaming.scheduler.JobScheduler logError - Error running job streaming job 1534871220000 ms.2
                org.apache.spark.SparkException: Job aborted due to stage failure: Task creation failed: java.io.FileNotFoundException: File does not exist: hdfs://mycluster/user/user1/sparkCheckpointData/2db59817-d954-41a7-9b9d-4ec874bc86de/rdd-1005/part-00000
                java.io.FileNotFoundException: File does not exist: hdfs://mycluster/user/user1/sparkCheckpointData/2db59817-d954-41a7-9b9d-4ec874bc86de/rdd-1005/part-00000
                        at org.apache.hadoop.hdfs.DistributedFileSystem$18.doCall(DistributedFileSystem.java:1122)
                at org.apache.hadoop.hdfs.DistributedFileSystem$18.doCall(DistributedFileSystem.java:1114)
                at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81)
                at org.apache.hadoop.hdfs.DistributedFileSystem.getFileStatus(DistributedFileSystem.java:1114)
                at org.apache.spark.rdd.ReliableCheckpointRDD.getPreferredLocations(ReliableCheckpointRDD.scala:89)
                at org.apache.spark.rdd.RDD$$anonfun$preferredLocations$1.apply(RDD.scala:273)
                at org.apache.spark.rdd.RDD$$anonfun$preferredLocations$1.apply(RDD.scala:273)
                at scala.Option.map(Option.scala:146)
                at org.apache.spark.rdd.RDD.preferredLocations(RDD.scala:273)
                at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$getPreferredLocsInternal(DAGScheduler.scala:1615)
                at org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$getPreferredLocsInternal$2$$anonfun$apply$1.apply$mcVI$sp(DAGScheduler.scala:1626)
                at org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$getPreferredLocsInternal$2$$anonfun$apply$1.apply(DAGScheduler.scala:1625)
                at org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$getPreferredLocsInternal$2$$anonfun$apply$1.apply(DAGScheduler.scala:1625)
                at scala.collection.immutable.List.foreach(List.scala:381)
                at org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$getPreferredLocsInternal$2.apply(DAGScheduler.scala:1625)
                at org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$getPreferredLocsInternal$2.apply(DAGScheduler.scala:1623)

进一步更新： 我发现 Spark 试图在 HDFS 中查找的 RDD 已被“ReliableRDDCheckpointData”进程删除，它为检查点数据创建了一个新的 RDD。 DAG 以某种方式指向这个旧的 RDD。如果对这些数据有任何引用，则不应将其删除。

Answer 1

考虑一下 Spark 流上的这种转换管道：

rtStream
                    .mapToPair(record -> new Tuple2<String, GenericDataModel>(record.key(), record.value()))                
                    .mapWithState(StateSpec.function(updateDataFunc).numPartitions(32)).stateSnapshots()                        
                    .foreachRDD(rdd -> {
                      if(counter ==1){
                       --convert RDD to Dataset, and register it as a SQL table names "InitialDataTable"
                      } else
                       --convert RDD to Dataset, and register it as a SQL table names "ActualDataTable"


                    }); 

mapWithState 与每批后状态数据的自动检查点相关联，因此上述“forEachRdd”块中的每个“rdd”都是检查点，并且在检查点时会覆盖之前的检查点（因为显然需要保持最新状态检查点）

但是让我们说如果用户仍在使用 rdd 编号 1，在我的情况下，我将第一个 rdd 注册为不同的表，并且每隔一个 rdd 作为不同的表，那么它不应该被覆盖。 （在java中也是一样的，如果某个对象引用了一个对象引用，那么该对象将不符合垃圾回收的条件）

现在，当我尝试访问“InitialDataTable”表时，显然用于创建该表的“rdd”已不在内存中，因此它将去 HDFS 从检查点恢复它，它不会找到它也在那里，因为它被下一个 rdd 覆盖，并且 spark 应用程序停止引用原因。

“org.apache.spark.SparkException：作业因阶段失败而中止：任务创建失败：java.io.FileNotFoundException：文件不存在：hdfs://mycluster/user/user1/sparkCheckpointData/2db59817-d954- 41a7-9b9d-4ec874bc86de/rdd-1005/part-00000"

所以要解决这个问题，我必须明确地检查第一个 rdd。