非法状态异常：压缩批次9时，_spark_metadata/0不存在

Question

我们有使用Spark结构化流实现的流应用程序，它试图从Kafka主题中读取数据，并将其写入HDFS位置。

有时应用程序会因异常而失败:

_spark_metadata/0 doesn't exist while compacting batch 9
java.lang.IllegalStateException: history/1523305060336/_spark_metadata/9.compact doesn't exist when compacting batch 19 (compactInterval: 10)

我们无法解决这个问题。

我找到的唯一解决方案是删除检查点位置文件，这将使作业在我们再次运行应用程序时从头开始读取主题/数据。然而，这对于生产应用程序来说不是一个可行的解决方案。

是否有人有针对此错误的解决方案，而无需删除检查点，以便我可以从上次运行失败的位置继续？

应用程序示例代码：

val df = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", <server list>)
  .option("subscribe", <topic>)
  .load()

[...] // do some processing

dfProcessed.writeStream
  .format("csv")
  .option("format", "append")
  .option("path",hdfsPath)
  .option("checkpointlocation","")
  .outputmode(append)
  .start

Answer 1

本文介绍了该机制，并提供了一种从Spark结构化流中已删除的_spark_metadata文件夹中恢复的好方法：

https://dev.to/kevinwallimann/how-to-recover-from-a-deleted-sparkmetadata-folder-546j

“创建虚拟日志文件：

如果元数据日志文件不可恢复，我们可以为丢失的微批处理创建虚拟日志文件。在我们的示例中，可以像这样完成：

for i in {0..1}; do echo v1 > "/tmp/destination/_spark_metadata/$i"; done

这将创建文件

/tmp/destination/_spark_metadata/0
/tmp/destination/_spark_metadata/1

现在，可以重新启动查询，并应在没有错误的情况下完成。“

因为我以前的输出文件夹无法再恢复。我尝试了这个虚拟解决方案，它可以摆脱非法状态例外：_spark_metadata/...不存在例外。

Answer 2

由< code>checkpointLocation引起的错误，因为< code>checkpointLocation存储旧的或已删除的数据信息。您只需删除包含< code>checkpointLocation的文件夹。

了解更多信息:https://kb . databricks . com/streaming/file-sink-streaming . html

例子：

df.writeStream
      .format("parquet")
      .outputMode("append")
      .option("checkpointLocation", "D:/path/dir/checkpointLocation")
      .option("path", "D:/path/dir/output")
      .trigger(Trigger.ProcessingTime("5 seconds"))
      .start()
      .awaitTermination()

您需要删除目录＜code＞checkpointLocation＜/code＞。

Answer 3

错误消息

_spark_metadata/n.compact doesn't exist when compacting batch n+10

可以在您

• 将一些数据处理到启用了检查点的FileSink中，然后
• 停止你的流媒体工作，然后
• 更改FileSink的输出目录，同时保持相同的检查点位置，然后
• 重启流作业

只需删除checkpointLocation中的文件并重新启动应用程序。

因为您不想删除检查点文件，所以您可以简单地将丢失的spark元数据文件从旧的文件接收器输出路径复制到新的输出路径。见下文了解什么是“缺失的spark元数据文件”。

为了理解为什么会抛出这个＜code＞IllegalStateException＜/code＞，我们需要了解所提供的文件输出路径中的幕后情况。让outPathBefore作为此路径的名称。当您的流作业正在运行并处理数据时，该作业将创建一个文件夹outPathBefore/_spark_metadata。在该文件夹中，您将找到一个以微批标识符命名的文件，其中包含数据已写入的文件（分区文件）列表，例如：

/home/mike/outPathBefore/_spark_metadata$ ls
0 1 2 3 4 5 6 7

在这种情况下，我们有8个微批次的详细信息。其中一个文件的内容如下所示

/home/mike/outPathBefore/_spark_metadata$ cat 0
v1
{"path":"file:///tmp/file/before/part-00000-99bdc705-70a2-410f-92ff-7ca9c369c58b-c000.csv","size":2287,"isDir":false,"modificationTime":1616075186000,"blockReplication":1,"blockSize":33554432,"action":"add"}

默认情况下，在每十个微批处理中，这些文件被压缩，这意味着文件0、1、2，...，9将存储在名为< code>9.compact的压缩文件中。

该过程连续进行随后的十个批次，即在微批次19中，作业聚合9.compact、10、11、12、…、19的最后10个文件。

现在，假设您让流式作业一直运行到微批处理15，这意味着该作业创建了以下文件:

/home/mike/outPathBefore/_spark_metadata/0
/home/mike/outPathBefore/_spark_metadata/1
...
/home/mike/outPathBefore/_spark_metadata/8
/home/mike/outPathBefore/_spark_metadata/9.compact
/home/mike/outPathBefore/_spark_metadata/10
...
/home/mike/outPathBefore/_spark_metadata/15

在第15个微批处理之后，您停止了流作业，并将文件接收器的输出路径更改为，比如说< code > outthafter 。由于您保持相同的检查点位置，流式作业将通过微批处理16继续。但是，它现在在新的输出路径中创建元数据文件:

/home/mike/outPathAfter/_spark_metadata/16
/home/mike/outPathAfter/_spark_metadata/17
...

现在，这是抛出异常的地方：当到达微批处理19时，作业试图压缩火花元数据文件夹中的第十个最新文件。但是，它只能找到文件16、17、18，但找不到9.compact、10等。因此错误消息如下：

java.lang.IllegalStateException: history/1523305060336/_spark_metadata/9.compact doesn't exist when compacting batch 19 (compactInterval: 10)

结构化流式编程指南解释了流式查询中更改后的恢复语义：

"不允许更改文件接收器的输出目录：sdf.writeStream.format（"parket"）. ption（"path"，"/xxPath"）到sdf.writeStream.format（"parket"）. ption（"path"，"/antherPath"）"

Databricks还在文章《使用文件接收器传输:更改检查点或输出目录时的恢复问题》中写了一些细节