Databricks中多数据源列名大小写不一致的兼容方案咨询

问题原因 #

Spark指定Schema读取CSV文件时，默认会严格匹配列名的大小写。所以当你用定义为orderdate的Schema读取Source1（列名是OrderDate）时，因大小写不匹配无法映射到对应字段，最终该列填充为NULL；反之用OrderDate的Schema读取Source2时也会出现同样问题。

解决方案 #

方案1：读取时忽略列名大小写（推荐） #

Spark的CSV读取器支持ignoreCase参数，开启后会忽略列名的大小写差异，自动匹配Schema中对应的字段。修改读取代码即可：

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DateType

# 定义目标Schema
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("orderdate", DateType(), True)
])

# 读取时添加ignoreCase参数，忽略列名大小写
df1 = spark.read.format("csv")\
    .option("header", "true")\
    .option("ignoreCase", "true")\
    .schema(schema)\
    .load("path/to/source1.csv")

df2 = spark.read.format("csv")\
    .option("header", "true")\
    .option("ignoreCase", "true")\
    .schema(schema)\
    .load("path/to/source2.csv")

# 合并数据
df = df1.union(df2)

方案2：读取后统一列名再合并 #

如果不想依赖ignoreCase参数，可以分别读取两个数据源后，将列名统一为目标Schema的名称，再进行合并：

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DateType

# 定义目标Schema
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("orderdate", DateType(), True)
])

# 读取Source1，并重命名OrderDate为orderdate
df1 = spark.read.format("csv")\
    .option("header", "true")\
    .load("path/to/source1.csv")\
    .withColumnRenamed("OrderDate", "orderdate")\
    .cast(schema)

# 读取Source2，直接匹配schema
df2 = spark.read.format("csv")\
    .option("header", "true")\
    .schema(schema)\
    .load("path/to/source2.csv")

# 合并数据
df = df1.union(df2)

方案3：先自动推断Schema，再统一列名 #

先不指定Schema读取文件，自动推断列名和类型，然后统一列名后转换为目标类型：

# 读取Source1，自动推断
df1 = spark.read.format("csv")\
    .option("header", "true")\
    .option("inferSchema", "true")\
    .load("path/to/source1.csv")\
    .withColumnRenamed("OrderDate", "orderdate")

# 读取Source2，自动推断
df2 = spark.read.format("csv")\
    .option("header", "true")\
    .option("inferSchema", "true")\
    .load("path/to/source2.csv")

# 统一Schema后合并（unionByName确保列名对应，不受顺序影响）
df = df1.unionByName(df2)
# 若需要强制转换为指定类型，可再调用cast方法
df = df.cast(schema)

注意事项 #

• 方案1的ignoreCase参数仅适用于CSV数据源，其他数据源（如Parquet）可能不支持，需根据实际情况选择。
• 使用unionByName替代union可以确保列名不对应时也能正确合并（Spark 2.3+支持），避免因列顺序不同导致的数据错误。

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