如何在Databricks中导入MongoDB表并处理大小写不敏感字段名？

问题原因：Databricks的Spark MongoDB连接器默认Schema推断为大小写不敏感，当MongoDB集合中存在大小写不同的同名字段（如account_id和ACCOUNT_ID）时，会被推断为同一列，导致写入时报COLUMN_ALREADY_EXISTS错误。

以下是三种可行的解决方案：

方案1：启用大小写敏感的Schema推断 #

在读取MongoDB数据时，添加spark.mongodb.input.schemaInference.caseSensitive选项并设置为true，让连接器严格区分字段名的大小写：

from cb_databricks.predefined import spark

mongo_df = (
    spark.read.format("mongo")
    .option("database", database)
    .option("collection", mongo_table)
    .option("spark.mongodb.input.uri", connect_string)
    .option("spark.mongodb.input.schemaInference.caseSensitive", "true")
    .load()
)
mongo_df.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("databricks_table")

该配置会将account_id和ACCOUNT_ID识别为两个独立列，避免列名冲突。

方案2：手动指定Schema（精准控制） #

如果需要更精确的列定义，或者大小写敏感推断仍存在问题，可以手动定义包含所有大小写字段的Schema，强制连接器按指定结构读取：

from cb_databricks.predefined import spark
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

# 自定义Schema，包含所有大小写混合的字段
custom_schema = StructType([
    StructField("_id", IntegerType(), nullable=False),
    StructField("account_id", IntegerType(), nullable=True),
    StructField("ACCOUNT_ID", IntegerType(), nullable=True),
    StructField("abc", StringType(), nullable=True),
    StructField("XYZ", StringType(), nullable=True)
])

mongo_df = (
    spark.read.format("mongo")
    .option("database", database)
    .option("collection", mongo_table)
    .option("spark.mongodb.input.uri", connect_string)
    .schema(custom_schema)
    .load()
)
mongo_df.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("databricks_table")

方案3：读取后重命名冲突字段 #

如果需要统一字段命名规则（如全部转小写），可以在读取数据后对冲突字段进行重命名处理：

from cb_databricks.predefined import spark

mongo_df = (
    spark.read.format("mongo")
    .option("database", database)
    .option("collection", mongo_table)
    .option("spark.mongodb.input.uri", connect_string)
    .load()
)

# 示例：将大写的ACCOUNT_ID重命名为account_id_upper，避免与小写字段冲突
for col in mongo_df.columns:
    if col == "ACCOUNT_ID":
        mongo_df = mongo_df.withColumnRenamed(col, "account_id_upper")

# 可选：统一将所有字段名转为小写（需确保无其他大小写冲突字段，避免数据丢失）
# mongo_df = mongo_df.toDF(*[col.lower() for col in mongo_df.columns])

mongo_df.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("databricks_table")

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