Databricks Delta Live Tables接入CSV时如何转换列名

核心处理逻辑是在cloud_files读取源数据后、写入DLT表之前，统一对所有列名执行规则清洗，不需要修改原始文件，同时完全保留增量加载的能力。

方案1：动态通用清洗（推荐，适配源端schema变更） #

如果源端CSV的列可能新增、调整，优先用DLT Python API实现，不需要提前硬编码列名，自动识别表头并按规则清洗，完整兼容Auto Loader的所有增量特性。
实现代码：

import dlt
import re
from pyspark.sql.functions import col

def clean_column_names(df):
    cleaned_columns = []
    for original_col in df.columns:
        # 1. 去除列名首尾空白字符
        processed_col = original_col.strip()
        # 2. 所有标点、特殊符号统一替换为下划线，保留字母、数字、下划线、中文字符
        processed_col = re.sub(r"[^\w\u4e00-\u9fa5]", "_", processed_col)
        # 3. 合并连续下划线，移除首尾残留下划线
        processed_col = re.sub(r"_+", "_", processed_col).strip("_")
        cleaned_columns.append(col(original_col).alias(processed_col))
    return df.select(*cleaned_columns)

@dlt.table(
    name="table_raw",
    comment="Ingesting cleaned data from /mnt/foo",
    table_properties={"quality": "bronze"}
)
def table_raw():
    raw_stream = spark.readStream.format("cloudFiles") \
        .option("cloudFiles.format", "csv") \
        .option("header", "true")
        # 可按需追加CSV配置，例如分隔符、字符集、schema演进模式
        # .option("delimiter", "\t")
        # .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .load("/mnt/foo/")
    return clean_column_names(raw_stream)

该方案的优势：

• 无硬编码列名，源端新增列会自动走清洗规则写入表，不需要手动修改代码
• 清洗规则可灵活扩展，比如统一列名大小写、重名列自动加后缀、关键字转义等逻辑都可以直接在clean_column_names函数中补充
• 完全保留Auto Loader的增量检测、schema推断、断点续传能力，和原SQL实现的性能一致

方案2：固定schema纯SQL实现 #

如果源端CSV的列完全固定不会变更，也可以直接在SQL中手动映射列别名完成清洗，示例代码：

CREATE INCREMENTAL LIVE TABLE table_raw
COMMENT "Ingesting cleaned data from /mnt/foo"
TBLPROPERTIES ("quality" = "bronze")
AS
SELECT
  -- 按原始列名逐个映射清洗后的别名，注意含特殊字符的原列名需要用反引号包裹
  `  Order ID,No.  ` AS order_id,
  `User Name!` AS user_name,
  `Amount($)` AS amount
  -- 其余列按相同规则补充映射
FROM cloud_files(
  "/mnt/foo/",
  "csv",
  map(
    "header", "true"
    -- 其他CSV配置可在此追加键值对
  )
)

注意：该方案无法自动适配源端列变更，一旦源端CSV调整表头、新增列，任务会直接报错，仅适合schema完全稳定的场景使用。

配置注意事项 #

• 读取CSV必须配置header = "true"参数，否则cloud_files会默认生成_c0/_c1格式的列名，无法获取原始表头做清洗
• 如果列名清洗后出现重名（例如原列名User Name和User!Name清洗后均为user_name），可以在Python清洗函数中增加重名检测逻辑，给重复列名追加数字后缀即可
• 若需要开启坏数据处理，直接在cloudFiles的option中增加对应坏数据路径配置即可，和列名清洗逻辑无冲突

内容的提问来源于stack exchange，提问作者Jared Dominic Caraan