使用Databricks Autoloader读取MS Defender湖屋数据遇重复列m如何解决？

问题原因 #

你的Blob存储路径包含两个同名分区键m=（分别对应月份和分钟），Autoloader会自动解析路径中的分区列，导致生成两个同名的m列。即使你在后续显式选择列，Spark在加载阶段就已经识别到重复列名，因此会触发歧义引用错误。

解决方案 #

方案1：禁用自动加载分区列 #

如果不需要分区列数据，可以直接通过cloudFiles.partitionColumns选项指定空值，阻止Autoloader加载任何分区列：

bronze_df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("badRecordsPath", bad_record_path)
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .option("cloudFiles.partitionColumns", "")  # 禁用分区列加载
  .load(f"{defender_adls}")
  .selectExpr("Tenant", "category", "operationName", "properties", "tenantId", "time")
)

方案2：自定义分区列Schema重命名重复列 #

如果需要保留分区列数据，可以通过cloudFiles.pathPartitionSchema显式定义分区列的结构，将重复的m分别重命名为不同的列名（比如month和minute）：

from pyspark.sql.types import StructType, StringType

# 定义分区列Schema，重命名重复的m键
partition_schema = StructType() \
    .add("y", StringType()) \
    .add("month", StringType())  # 对应路径中的第一个m=（月份）
    .add("d", StringType()) \
    .add("h", StringType()) \
    .add("minute", StringType())  # 对应路径中的第二个m=（分钟）

bronze_df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("badRecordsPath", bad_record_path)
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .option("cloudFiles.pathPartitionSchema", partition_schema.json())  # 指定自定义分区Schema
  .load(f"{defender_adls}")
  .selectExpr("Tenant", "category", "operationName", "properties", "tenantId", "time")
)

为什么显式选列无效？ #

Spark在执行load操作时会先解析所有列（包括自动识别的分区列），此时DataFrame中已经存在两个m列。即使后续选择其他列，Spark在解析元数据阶段就会检测到列名冲突，因此报错会在selectExpr执行前触发。

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