如何通过Delta Lake修改列名并获取旧列名以更新.ttl映射文件？

一、物理重写数据文件实现列名修改（适配Dremio） #

由于列映射仅修改元数据，无法被Dremio识别，因此需要直接重写数据文件，让新列名在物理层面生效：

方法1：Spark DataFrame覆盖写入 #

// 读取原Delta表
val originalDF = spark.read.format("delta").load("/path/to/your/table")

// 重命名目标列
val renamedDF = originalDF
  .withColumnRenamed("creditcard_number", "number")
  .withColumnRenamed("creditcard_type", "type")

// 覆盖写入原表，开启overwriteSchema确保Schema更新
renamedDF.write
  .format("delta")
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .save("/path/to/your/table")

方法2：Spark SQL方式 #

-- 创建临时视图映射新列名
CREATE OR REPLACE TEMP VIEW renamed_view AS
SELECT 
  `customer IDs`,
  creditcard_number AS number,
  creditcard_type AS type
FROM your_table;

-- 覆盖写入原表
CREATE OR REPLACE TABLE your_table
USING DELTA
AS SELECT * FROM renamed_view;

操作完成后，数据文件的列名将被物理替换，Dremio可正常识别新列名。

二、留存旧列名的可访问方案 #

方案1：存储在Delta表自定义属性中 #

将列名映射写入表的自定义属性，方便后续快速查询：

ALTER TABLE your_table SET TBLPROPERTIES (
  'column.mapping.old_new' = 'creditcard_number:number, creditcard_type:type'
);

查询属性获取映射关系：

-- 查看表的完整属性
DESCRIBE EXTENDED your_table;

-- 用Spark精准提取属性
val props = spark.sql("DESCRIBE EXTENDED your_table")
  .filter($"col_name" === "Properties")
  .select($"data_type")
  .head()
  .getString(0)

方案2：创建独立的列变更日志表 #

专门维护一张日志表记录所有列变更历史，适合批量更新.ttl文件的场景：

-- 创建列变更日志表
CREATE TABLE IF NOT EXISTS column_change_log (
  table_name STRING,
  old_column_name STRING,
  new_column_name STRING,
  change_time TIMESTAMP
) USING DELTA;

-- 插入本次列变更记录
INSERT INTO column_change_log
VALUES
('your_table', 'creditcard_number', 'number', CURRENT_TIMESTAMP()),
('your_table', 'creditcard_type', 'type', CURRENT_TIMESTAMP());

后续可直接查询该表获取所有列名变更记录，用于自动替换.ttl文件中的旧列名。

三、Delta事件日志的列变更支持 #

Delta Lake的_delta_log目录下的JSON日志会记录所有表结构变更操作，包括列重命名。你可以直接解析这些日志提取旧列名与新列名的映射：

1. 对应变更的日志文件中，commitInfo字段会包含操作详情：

{
  "commitInfo": {
    "timestamp": 1700000000000,
    "operation": "ALTER TABLE",
    "operationParameters": {
      "renameColumns": "{\"creditcard_number\":\"number\",\"creditcard_type\":\"type\"}"
    }
  }
}

2. 用Spark批量解析日志：

val deltaLogDF = spark.read.json("/path/to/your/table/_delta_log/*.json")
val renameOps = deltaLogDF
  .filter($"commitInfo.operation" === "ALTER TABLE")
  .select($"commitInfo.operationParameters.renameColumns")

注意：变更数据馈送（CDF）仅跟踪行级数据变更，不包含列级元数据变更，需直接解析事件日志获取相关信息。

你之前遗漏的关键方法 #

1. 放弃列映射的元数据变更方式，采用物理重写数据文件的方式修改列名，适配Dremio的识别逻辑；
2. 利用Delta表的自定义属性或独立日志表留存列名映射关系；
3. 直接解析Delta事件日志获取列变更历史，而非依赖CDF。

内容的提问来源于stack exchange，提问作者Roland R.