Redshift Spectrum与Delta Lake集成更新分区列的分区表异常问题

问题根源 #

Redshift Spectrum依赖AWS Glue Data Catalog中的分区元数据执行查询。当Delta表删除旧分区并重新生成Symlink Manifest时，Glue Catalog中的旧分区元数据未同步删除，Redshift会尝试访问已被Delta清理的旧分区Manifest路径，触发S3 404错误。而Athena会自动校验路径有效性，因此不会出现同类问题。

自动化解决方法 #

1. 同步Delta分区与Glue Catalog元数据 #

在生成Symlink Manifest的Glue任务后，添加步骤自动清理Glue Catalog中的无效分区：

步骤1：获取Delta表当前有效分区

在Scala代码中读取Delta表的现存分区列表：

import io.delta.tables.DeltaTable

val deltaTablePath = "s3://xxxxxxx/v0/data/"
val deltaTable = DeltaTable.forPath(spark, deltaTablePath)

// 查询当前所有存在的分区
val existingPartitions = spark.sql(
  s"SELECT DISTINCT year, month, day FROM delta.`$deltaTablePath`"
).collect().map(row => (row.getString(0), row.getString(1), row.getString(2)))

步骤2：获取Glue Catalog中的分区

通过Redshift查询外部表的分区元数据：

SELECT year, month, day 
FROM SVV_EXTERNAL_PARTITIONS 
WHERE schemaname = 'yyyyy' AND tablename = 'xxxxxxxx'

将结果转换为(year, month, day)元组列表。

步骤3：对比并删除无效分区

找出Glue中存在但Delta已删除的分区，生成并执行DROP PARTITION语句：

import java.sql.DriverManager

// 建立Redshift JDBC连接（需替换为实际参数）
val conn = DriverManager.getConnection("jdbc:redshift://<redshift-endpoint>:5439/<db-name>?user=<user>&password=<password>")
val stmt = conn.createStatement()

// gluePartitions为从Redshift查询得到的分区列表
val gluePartitions = // 转换为(year, month, day)元组的列表

// 计算需要删除的分区
val partitionsToDrop = gluePartitions.diff(existingPartitions)

partitionsToDrop.foreach { case (year, month, day) =>
  val dropSql = s"ALTER TABLE yyyyy.xxxxxxxx DROP PARTITION (year='$year', month='$month', day='$day');"
  stmt.execute(dropSql)
}

stmt.close()
conn.close()

2. 生成空Manifest文件规避404 #

若不想修改Glue元数据，可在Delta删除分区后，自动在旧分区路径下创建空的manifest文件，避免Redshift报错：

import com.amazonaws.services.s3.AmazonS3ClientBuilder
import com.amazonaws.services.s3.model.PutObjectRequest

val s3Client = AmazonS3ClientBuilder.defaultClient()
val bucketName = "xxxxxxx"

partitionsToDrop.foreach { case (year, month, day) =>
  val manifestPath = s"v0/data/_symlink_format_manifest/year=$year/month=$month/day=$day/manifest"
  // 上传空文件到对应路径
  s3Client.putObject(new PutObjectRequest(bucketName, manifestPath, new java.io.File("/dev/null")))
}

3. 升级Delta Lake版本（长期推荐） #

Delta Lake 2.0+版本优化了与Redshift Spectrum的集成，新增了自动清理旧Manifest分区的配置。若条件允许，升级至较新版本后，可通过设置delta.symlinkFormatManifest.cleanup.enabled = true自动清理无效Manifest文件及关联元数据。

总结 #

• 针对Delta 1.0.1版本，优先选择同步分区元数据或生成空Manifest文件实现自动化处理；
• 长期运维建议升级Delta Lake至2.0+版本，利用官方内置的自动清理能力降低运维成本。

内容的提问来源于stack exchange，提问作者Antonio La Macchia