如何高效获取Delta Lake写入操作中的分区值？

在用Spark+Delta Lake写入超大规模分区数据集时，需要快速拿到本次写入涉及的所有分区值（数据量极大、分区数多，遍历全量数据的方式完全没法用）。

写入代码示例：

data.write.format("delta").partitionBy("DATE", "GENRE").save()

期望结果格式：

DATE=2024-03-01/GENRE=Female
DATE=2024-03-01/GENRE=Male
....

目前采用的遍历全量数据存入累加器的方式效率极低：

val partitionCols = Seq("DATE", "GENRE") // Multiple partitions
dataset.mapPartitions { partition =>
  partition.map { row =>
    val partitionValues = partitionCols.map { partition =>
        (partition, Option(row.get(row.fieldIndex(partition))).getOrElse("").toString)
    }
    setAccumulator.add(partitionValues)
    row
  }
}

试过几个方案但都有疑问：

• 从指定版本/时间戳加载DeltaLog遍历操作：不确定效率与可靠性
• 启用delta.checkpoint.writeStatsAsStruct获取partitionValues_parsed：不清楚checkpoint的开销及数据覆盖情况
• 使用readChangeFeed读取变更数据：无法确保仅获取本次写入数据，仍需遍历新增数据
• 使用_symlink_format_manifest：需处理历史文件夹问题

已知Delta写入过程中会记录受影响的分区值，但找不到高效获取方式，了解PostCommitHook但不清楚配置方法，求高效解决方案。

一、使用PostCommitHook（首推） #

Delta Lake的PostCommitHook是写入完成后自动触发的钩子，直接从提交的元数据中提取本次写入的分区，无需接触原始数据，效率拉满。

具体操作： #

1. 自定义Hook类，继承PostCommitHook并重写run方法：

import org.apache.spark.sql.delta.hooks.PostCommitHook
import org.apache.spark.sql.delta.{DeltaLog, Snapshot}
import org.apache.spark.sql.SparkSession

class CaptureWrittenPartitionsHook extends PostCommitHook {
  override def run(
      spark: SparkSession,
      deltaLog: DeltaLog,
      commitVersion: Long,
      snapshot: Snapshot,
      operationMetrics: Map[String, String]): Unit = {
    // 获取本次提交新增的文件，提取分区值
    val addedFiles = snapshot.deltas.last.addedFiles
    val partitionCols = deltaLog.snapshot.metadata.partitionColumns
    val writtenPartitions = addedFiles
      .map(_.partitionValues)
      .distinct
      .map { kv =>
        partitionCols.map(col => s"$col=${kv(col)}").mkString("/")
      }
    
    // 这里可将分区值输出到日志、存储系统或通过累加器传递
    println("本次写入涉及的分区：")
    writtenPartitions.foreach(println)
  }
}

2. 配置Spark启用该Hook：
   写入前添加配置：

spark.conf.set("delta.postCommitHooks", "com.yourpackage.CaptureWrittenPartitionsHook")

若需全局生效，可将配置加入spark-defaults.conf：

delta.postCommitHooks com.yourpackage.CaptureWrittenPartitionsHook

优势： #

• 直接读取Delta提交的元数据（小文件），完全不处理原始数据，速度极快
• 精准获取本次写入的分区，不会混入历史数据

二、写入后读取DeltaLog最新版本（无需修改写入逻辑） #

若无法使用Hook，可在写入完成后直接读取DeltaLog的最新提交版本，提取新增文件的分区值。

代码示例： #

import org.apache.spark.sql.delta.DeltaLog
import org.apache.spark.sql.SparkSession

val spark = SparkSession.active
val deltaTablePath = "/path/to/your/delta/table"
val deltaLog = DeltaLog.forTable(spark, deltaTablePath)
// 获取最新提交版本
val latestVersion = deltaLog.snapshot.version
// 获取该版本新增的所有文件
val addedFiles = deltaLog.getChanges(latestVersion, latestVersion).flatMap(_.addFiles)
// 格式化分区值
val partitionCols = deltaLog.snapshot.metadata.partitionColumns
val writtenPartitions = addedFiles
  .map(_.partitionValues)
  .distinct
  .map(kv => partitionCols.map(col => s"$col=${kv(col)}").mkString("/"))

// 输出结果
writtenPartitions.foreach(println)

注意事项： #

• 务必等写入操作完全完成后再执行该逻辑，避免读取到未提交的版本
• 效率远高于遍历全量数据，仅读取Delta的元数据文件，数据量极小

三、利用Checkpoint的结构化统计（备选） #

启用delta.checkpoint.writeStatsAsStruct后，Checkpoint会包含partitionValues_parsed字段，可直接提取分区信息，但需注意Checkpoint的生成时机。

操作步骤： #

1. 启用配置：

spark.conf.set("delta.checkpoint.writeStatsAsStruct", "true")

2. 写入完成后读取最新Checkpoint文件：

import org.apache.spark.sql.functions._

val deltaPath = "/path/to/your/delta/table"
// 获取最新Checkpoint版本
val checkpointVersion = spark.read.text(s"$deltaPath/_delta_log/_last_checkpoint")
  .head().getString(0).toLong
// 读取Checkpoint的Parquet文件
val checkpointDF = spark.read.parquet(s"$deltaPath/_delta_log/$checkpointVersion.checkpoint.parquet")

// 提取并格式化分区值
val partitionCols = deltaLog.snapshot.metadata.partitionColumns
val writtenPartitions = checkpointDF
  .select(explode(col("partitionValues_parsed")).alias("partition"))
  .select("partition.*")
  .distinct()
  .map(row => partitionCols.map(col => s"$col=${row.getAs[String](col)}").mkString("/"))
  .collect()

writtenPartitions.foreach(println)

优缺点： #

• 优点：Checkpoint为结构化数据，解析方便
• 缺点：Checkpoint默认每10次提交生成一次，若写入后未生成需手动触发deltaLog.checkpoint()，会产生额外开销

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