Spark Streaming无法恢复挂起的批处理TextFileStream

如何解决Spark Streaming无法恢复挂起的批处理TextFileStream

我正在使用Spark 2.3.1来处理文本文件流。上下文设置如下：

val ssc = new StreamingContext(sparkSession.sparkContext,Seconds(config.sparkStreamingBatchTime))
ssc.checkpoint(config.sparkCheckpointLocation)

一切正常，除非没有为作业提供足够的资源并且批处理开始堆积。解决方案是终止工作，提供更多资源，然后重新开始。

不幸的是，这样做时，已触发但尚未处理的待处理批次会丢失。

示例：我每分钟安排一次作业，仅执行Thread.sleep(3 * 60 * 1000)（三分钟）。批次开始按预期堆积。

为批次引入了一个新文件：

20/09/23 08:39:00 DEBUG FileInputDStream: Getting new files for time 1600850340000,ignoring files older than 1600850280000
20/09/23 08:39:00 DEBUG FileInputDStream: hdfs://nameservice1/[...]/test_1 accepted with mod time 1600850308293
20/09/23 08:39:00 INFO org.apache.spark.streaming.dstream.FileInputDStream: Finding new files took 18 ms

随着文件集的保存，检查点系统显然可以正常工作：

20/09/23 08:39:00 DEBUG FileInputDStream: Updated checkpoint data for time 1600850340000 ms: [
3 file sets
1600850280000 ms ->
1600850340000 ms -> hdfs://nameservice1/[...]/test_1
1600850220000 ms ->
]

然后，我杀死了该工作（由于调用了shudown挂钩，因此很优雅）：

20/09/23 08:40:00 INFO JobScheduler: Stopped JobScheduler
20/09/23 08:40:00 INFO StreamingContext: StreamingContext stopped successfully
20/09/23 08:40:00 INFO SparkContext: Invoking stop() from shutdown hook
[...]
20/09/23 08:40:00 INFO org.apache.spark.util.ShutdownHookManager: Shutdown hook called

最后，当我再次启动该作业时，由于时间戳，该文件将被忽略，但首先未对其进行处理！

20/09/23 08:44:00 DEBUG FileInputDStream: Getting new files for time 1600850640000,ignoring files older than 1600850580000
20/09/23 08:44:00 DEBUGFileInputDStream: hdfs://nameservice1/[...]/test_1 ignored as mod time 1600850308293 <= ignore time 1600850580000
20/09/23 08:44:00 INFO FileInputDStream: Finding new files took 3 ms

有一种解决方法涉及“触摸”或复制未处理的文件，但这是在生产中进行维护的噩梦。您是否认为我在关机时丢失了某些东西，因此待处理的批处理仍保留在检查点文件夹中？是否可以覆盖时间戳检查？还有其他想法吗？

谢谢！

2020年9月24日更新：我有一个解决方法，但确实很难看。基于this question，我设法获得了生成每批提供的RDD的文件：

/**
  * Recursive method to extract original Metadata files involved in this batch.
  * @param rdd Each RDD created for each batch.
  * @return All HDFS files originally read.
  */
def extractSourceHDFSFiles(rdd: RDD[_]): Set[String] = {
  def extractSourceHDFSFilesWithAcc(rdd: List[RDD[_]]) : Set[String] = {
    rdd match {
      case Nil => Set()
      case head :: tail => {
        val name = head.toString()
        if (name.startsWith("hdfs")){
          Set(name.split(" ")(0)) ++ extractSourceHDFSFilesWithAcc(head.dependencies.map(_.rdd).toList) ++ extractSourceHDFSFilesWithAcc(tail)
        }
        else {
          extractSourceHDFSFilesWithAcc(
head.dependencies.map(_.rdd).toList) 
++ extractSourceHDFSFilesWithAcc(tail)
        }
      }
    }
}

至少，我可以使用它记录已处理的文件。然后，手动搜索未处理的内容。这很糟糕，但这是我唯一的工作经验。

更新＃2 09/24/2020：我注意到我可以使用--conf spark.streaming.fileStream.minRememberDuration使Spark考虑较早的元数据文件，但这不会阻止它重新处理已经处理过的文件。我将需要使用先前更新中的信息来删除这些文件，并迫使管理员在启动作业之前清理旧的元数据文件...

更新＃3：2020年9月30日我已经研究了FileInputDStream的源代码，并且对检查点的功能有了更好的了解。它可以恢复被监视目录的“快照”以及恢复在什么时间出现的文件。

但是，这里缺少的部分是已生成但从未处理过的那些“文件集”（即批处理），如何在重新启动工作后将其恢复？

20/09/23 08:39:00 DEBUG FileInputDStream: Updated checkpoint data for time 1600850340000 ms: [
3 file sets
1600850280000 ms ->
1600850340000 ms -> hdfs://nameservice1/[...]/test_1
1600850220000 ms ->
]