Spark查询中Parquet文件覆盖时如何保障原子性与数据完整性以优化压缩？

针对现有Spark环境中Parquet分区持续查询、后台优化作业需原子覆盖文件的场景，以下是无需依赖数据湖仓的可行方案：

方案1：临时目录+原子重命名 #

• 操作逻辑：后台优化作业先将处理后的Parquet文件写入同一存储系统下的独立临时目录（例如原分区路径的兄弟目录/data/table/part=xxx_temp），待所有文件写入完成且校验通过后，执行原子重命名操作替换原分区目录。
• 原子性原理：多数分布式存储（HDFS、S3等）的rename操作具备原子性——要么完全替换成功，原目录被新目录替代；要么替换失败，原目录保持不变。正在运行的查询会持续读取原目录直到重命名完成，后续新查询自动读取新目录，不会出现中间状态的脏数据。
• 关键注意点：
  • 临时目录需配置自动清理机制，避免作业失败后残留垃圾文件。
  • 重命名前必须校验临时目录的文件完整性（如对比文件数量、计算校验和），防止写入未完成就替换原数据。
  • 针对S3这类对象存储，rename本质是复制+删除，但可通过批量操作结合存储原生版本控制辅助降低风险。

方案2：分区级版本切换 #

• 操作逻辑：给原分区路径添加版本标识，例如原分区为/data/table/part=xxx/v1，优化作业写入/data/table/part=xxx/v2，随后通过修改软链接将分区的访问入口指向新版本目录。
• 原子性原理：软链接的更新操作是原子的，查询通过软链接访问数据，切换瞬间完成，不会出现数据断裂或混合读取的情况。
• 关键注意点：
  • 需确保Spark查询配置支持跟随软链接（如HDFS需调整dfs.client.read.shortcircuit.skip.checksum等参数，具体依存储系统而定）。
  • 旧版本目录需延迟删除，确认所有基于旧数据的查询完成后再清理，避免运行中查询报错。
  • 对于不支持软链接的对象存储（如S3），可通过内部路由层或存储网关实现前缀映射的版本切换。

方案3：Spark原生原子提交协议 #

• 操作逻辑：利用Spark内置的原子提交协议，将优化后的文件先写入临时文件，待全部写入完成后一次性替换原分区文件。
• 具体配置：
  • 针对HDFS，使用org.apache.spark.sql.execution.datasources.HadoopFsCommitProtocol，该协议会先将文件写入.tmp临时目录，提交阶段原子移动至目标路径。
  • 代码示例：

    spark.conf.set("spark.sql.sources.commitProtocolClass", "org.apache.spark.sql.execution.datasources.HadoopFsCommitProtocol")
    df.write.mode("overwrite").partitionBy("part").parquet("/data/table")
    

• 关键注意点：
  • 该方案适用于全分区覆盖场景，若需增量优化部分文件，需配合文件级的原子替换逻辑。
  • 需配置作业唯一标识作为临时目录前缀，避免多作业重试导致的文件冲突。

方案4：读写分离+延迟删除 #

• 操作逻辑：优化作业将新生成的Parquet文件写入并行的新分区路径（如/data/table_new/part=xxx），待写入完成后，将查询流量切换至新路径，再延迟删除旧分区文件。
• 原子性原理：流量切换是一次性操作，查询要么读取旧数据，要么读取新数据，不会出现混合读取的中间状态。
• 关键注意点：
  • 可通过Spark SQL视图统一数据访问入口，切换时仅需修改视图的底层路径，无需调整所有查询代码。
  • 旧数据删除需预留足够长的延迟时间，确保所有正在运行的旧查询已执行完成，避免查询过程中文件被删除引发报错。

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