合并ADLS中不同文件夹Parquet文件时遇transpose尺寸匹配异常求助

问题场景 #

在ADLS存储的不同文件夹下有多个Parquet文件，使用循环union合并时触发错误：

IllegalArgumentException: transpose requires all collections have the same size

原代码如下：

files = dbutils.fs.ls('abfss://udl-container@container-name.dfs.core.windows.net/UserData/folder1/folder2/folder3/')

combined_df = None
for fi in files: 
    df = spark.read.parquet(fi.path)
    if combined_df == None:
       combined_df = df
    else:
      combined_df = combined_df.union(df)

错误原因 #

这个错误的核心是待合并的DataFrame列结构不匹配：

• Spark原生的union方法要求两个DF的列数量、顺序、数据类型完全一致，它会按列的位置而非名称进行匹配
• 如果某几个Parquet文件的列数不同、列顺序不一致，或者同一列名对应的数据类型有差异，就会触发转置操作失败的报错

解决方案 #

方案1：直接读取整个目录（最推荐） #

Spark本身支持直接读取目录下所有Parquet文件，会自动按列名对齐结构，并行处理效率远高于循环读取，代码极简：

combined_df = spark.read.parquet(
    'abfss://udl-container@container-name.dfs.core.windows.net/UserData/folder1/folder2/folder3/'
)
# 写入合并后的文件（按需调整模式和路径）
combined_df.write.mode("overwrite").parquet("合并后文件存储路径")

方案2：循环时改用unionByName（需逐个处理文件时使用） #

如果必须对每个文件做额外处理（比如过滤、日志记录），把union替换为unionByName，它会按列名匹配而非位置，还支持允许缺失列：

files = dbutils.fs.ls('abfss://udl-container@container-name.dfs.core.windows.net/UserData/folder1/folder2/folder3/')

combined_df = None
for fi in files: 
    df = spark.read.parquet(fi.path)
    if combined_df is None:
       combined_df = df
    else:
      # allowMissingColumns=True 允许部分文件缺失列，缺失列自动填充Null
      combined_df = combined_df.unionByName(df, allowMissingColumns=True)

方案3：强制统一Schema（严格控制列结构） #

如果需要确保合并后的DF严格符合指定列结构，可以提前定义基准Schema，读取每个文件时强制应用：

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

# 根据实际业务需求定义基准Schema
base_schema = StructType([
    StructField("user_id", IntegerType(), nullable=True),
    StructField("user_name", StringType(), nullable=True),
    StructField("register_time", StringType(), nullable=True)
])

combined_df = None
files = dbutils.fs.ls('abfss://udl-container@container-name.dfs.core.windows.net/UserData/folder1/folder2/folder3/')
for fi in files: 
    # 读取时强制使用基准Schema，确保结构统一
    df = spark.read.schema(base_schema).parquet(fi.path)
    if combined_df is None:
       combined_df = df
    else:
      combined_df = combined_df.unionByName(df)

注意事项 #

• 尽量避免用dbutils.fs.ls循环读取文件，Spark的目录读取机制更高效且能自动处理大部分结构不一致问题
• 可以先排查异常文件：通过读取单个文件查看Schema，定位哪些文件的列结构和其他文件不同

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