使用Databricks Autoloader读取CSV时，如何处理列名无效字符？

你遇到的错误是因为Delta Lake默认不允许列名包含空格、逗号这类特殊字符，按照提示启用列映射功能即可解决，核心是设置表属性delta.columnMapping.mode为name，具体实现分两种场景：

场景一：首次创建Delta表（推荐） #

你可以直接在流写入时指定表属性，或者先通过SQL预创建带属性的表：

方法1：流写入时直接配置属性 #

在你的读取流代码之后，添加写入逻辑并指定列映射属性：

# 你的读取流代码（保持不变）
stream = spark.readStream.format("cloudFiles")\
        .option('cloudFiles.format', 'csv')\
        .option('cloudFiles.schemaLocation', delta_loc)\
        .option("rescuedDataColumn", "_rescued_data")\
        .option('header', 'true')\
        .option('delimiter', '|')\
        .option('pathGlobFilter', f"*{file_code}*.csv")\
        .load(data_path)

# 写入Delta表时配置列映射属性
stream.writeStream \
  .format("delta") \
  .option("checkpointLocation", "/path/to/checkpoint/dir")  # 替换为你的检查点路径
  .option("mergeSchema", "true") \
  .option('delta.columnMapping.mode', 'name') \
  .option('delta.minReaderVersion', '5') \
  .option('delta.minWriterVersion', '5') \
  .start("/path/to/delta/table/location")  # 替换为你的Delta表存储路径

方法2：先通过SQL创建带属性的表 #

先执行SQL语句预定义表结构和属性，再写入流数据：

CREATE TABLE IF NOT EXISTS your_target_table
USING DELTA
LOCATION '/path/to/delta/table'
TBLPROPERTIES (
  'delta.columnMapping.mode' = 'name',
  'delta.minReaderVersion' = '5',
  'delta.minWriterVersion' = '5'
)

然后写入流：

stream.writeStream \
  .format("delta") \
  .option("checkpointLocation", "/path/to/checkpoint/dir") \
  .option("mergeSchema", "true") \
  .table("your_target_table")

场景二：已有Delta表需要修改属性 #

如果已经存在未启用列映射的Delta表，先执行SQL修改表属性：

ALTER TABLE your_target_table SET TBLPROPERTIES (
  'delta.columnMapping.mode' = 'name',
  'delta.minReaderVersion' = '5',
  'delta.minWriterVersion' = '5'
)

修改完成后，重启你的流作业即可生效。

注意事项 #

• 启用列映射后，查询含空格的列名时需要用反引号包裹，例如：SELECT User Name FROM your_target_table
• delta.minReaderVersion和delta.minWriterVersion必须设置为5，这是启用列映射的最低版本要求

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