EMR上Structured Streaming写入文件Sink失败：/_spark_metadata/9.compact不存在

嘿，我在EMR上用早期Spark版本做Structured Streaming的时候也踩过这个坑——Console Sink能正常跑，但File Sink动不动就报元数据文件找不到，本质上是Spark 2.2.x的流处理元数据机制和S3/EMRFS的一致性特性不兼容导致的，咱们一步步来搞定它：

问题根源 #

Spark 2.2.x的File Sink会在输出目录下生成_spark_metadata目录来管理流状态和文件元数据，其中的compact文件是用来合并日志的。但S3是最终一致性存储，加上早期EMRFS的一致性保障不够完善，再配合Spark对S3的适配还存在小bug，就会出现Spark尝试读取还没完全同步的compact文件，或者compact过程中因一致性问题导致文件丢失的情况。尤其如果把checkpoint目录直接放在S3上，这个问题会更严重。

解决方案 #

1. 把Checkpoint目录移到HDFS（关键操作！） #

S3的最终一致性是核心诱因，所以一定要把checkpoint目录放到EMR自带的HDFS（强一致性存储），而非S3。同时调整几个参数减少compact操作的频率：
启动spark-shell时可以直接加这些配置：

spark-shell --conf spark.sql.streaming.checkpointLocation=hdfs:///user/spark/checkpoint/event_stream \
            --conf spark.sql.streaming.fileSink.log.compactInterval=100 \
            --conf spark.sql.streaming.fileSink.log.cleanupDelay=3600000 \
            --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem

或者在代码里动态设置：

spark.conf.set("spark.sql.streaming.checkpointLocation", "hdfs:///user/spark/checkpoint/event_stream")
spark.conf.set("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")

参数说明：

• compactInterval：默认是10，调大到100可以减少compact操作的次数，降低出错概率
• S3AFileSystem：EMR 5.x里S3A比老的S3FileSystem对Structured Streaming的支持更稳定

2. 清理损坏的元数据目录 #

如果已经出现报错，先删除S3输出目录下的_spark_metadata文件夹，然后重启流作业。注意：这会让作业从头开始消费Kafka数据，如果需要断点续传，可以先通过Kafka的offset管理工具记录当前消费的offset，重启时指定startingOffsets参数。

3. 优化EMRFS一致性配置 #

在EMR集群的core-site.xml里添加或修改以下配置，增强EMRFS的一致性保障：

<property>
  <name>fs.s3.consistent</name>
  <value>true</value>
</property>
<property>
  <name>fs.s3.consistent.retryPolicyType</name>
  <value>exponential</value>
</property>

这会让EMRFS在读写S3时做一致性检查，减少因S3最终一致性导致的文件找不到问题。

修改后的完整代码示例 #

// 先配置必要参数
spark.conf.set("spark.sql.streaming.checkpointLocation", "hdfs:///user/spark/checkpoint/event_stream")
spark.conf.set("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")

// 读取Kafka数据
val event = spark.readStream.format("kafka")
  .option("kafka.bootstrap.servers", "<your-server-list>")
  .option("subscribe", "<your-topic>")
  .load()

// 解析JSON数据（替换成你的实际Schema）
import org.apache.spark.sql.types._
val eventSchema = StructType(Seq(
  StructField("id", StringType),
  StructField("timestamp", TimestampType),
  StructField("content", StringType)
))

val eventdf = event.select($"value".cast("string").as("json"))
  .select(from_json($"json", eventSchema).as("data"))
  .select("data.*")

// 写入S3的Parquet文件
val query = eventdf.writeStream
  .format("parquet")
  .option("path", "s3a://your-bucket/path/to/output")
  .option("checkpointLocation", "hdfs:///user/spark/checkpoint/event_stream") // 再次明确指定checkpoint到HDFS
  .start()

query.awaitTermination()

额外建议 #

如果条件允许，尽量升级到EMR 5.20+版本（对应Spark 2.4.x），这个版本修复了很多Structured Streaming和S3交互的bug，稳定性会提升很多。

内容的提问来源于stack exchange，提问作者maverik