PySpark多格式文件读取逻辑构建及空DataFrame创建问题

原代码中的核心问题 #

• 创建空DataFrame时，schema参数需要传入Spark StructType对象，而非普通列表；原代码里[a,b,c,d]的变量未定义也未加引号，完全不符合要求。
• 条件判断误用赋值符号=，正确应该用比较符号==。
• 未从file变量中提取文件格式，format、CSVFile、ParquetFile都是未定义的变量。
• spar、parque是拼写错误，正确应为spark、parquet。
• union操作逻辑错误：df1.union(df1)是将DataFrame与自身合并，而非合并到finalDF；且空DataFrame与非空DataFrame合并时，必须保证schema完全一致。
• 循环中反复执行union会降低性能，建议先收集所有DataFrame再一次性合并。

修正后的代码示例 #

1. 先定义正确的Schema #

首先要明确列名和对应的数据类型，示例如下：

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

# 定义schema：列名+对应数据类型
target_schema = StructType([
    StructField("a", StringType(), nullable=True),
    StructField("b", IntegerType(), nullable=True),
    StructField("c", StringType(), nullable=True),
    StructField("d", IntegerType(), nullable=True)
])

2. 遍历文件并合并DataFrame #

# 初始化空DataFrame
finalDF = spark.createDataFrame([], schema=target_schema)

list_of_files = ["file1.csv", "file2.parquet", "file3.csv"]  # 替换为你的实际文件列表

for file in list_of_files:
    # 提取文件后缀
    file_ext = file.split(".")[-1].lower()
    
    if file_ext == "csv":
        # 读取CSV时指定schema，保证和finalDF一致
        # 若CSV文件带表头，将header参数设为True
        df = spark.read.csv(file, schema=target_schema, header=False)
        finalDF = finalDF.union(df)
    elif file_ext == "parquet":
        # 读取Parquet文件，Parquet自带schema，需确保和目标schema匹配
        df = spark.read.parquet(file)
        # 若Parquet schema与目标不一致，手动调整列和数据类型
        df = df.select("a", "b", "c", "d") \
               .withColumn("b", df["b"].cast(IntegerType())) \
               .withColumn("d", df["d"].cast(IntegerType()))
        finalDF = finalDF.union(df)

finalDF.show()

3. 更高效的合并方式（避免循环union） #

循环中反复执行union会触发多次计算，更优方式是先收集所有读取的DataFrame，再用reduce一次性合并：

from functools import reduce
from pyspark.sql import DataFrame

dfs = []
for file in list_of_files:
    file_ext = file.split(".")[-1].lower()
    if file_ext == "csv":
        df = spark.read.csv(file, schema=target_schema, header=False)
        dfs.append(df)
    elif file_ext == "parquet":
        df = spark.read.parquet(file)
        df = df.select("a", "b", "c", "d") \
               .withColumn("b", df["b"].cast(IntegerType())) \
               .withColumn("d", df["d"].cast(IntegerType()))
        dfs.append(df)

# 合并所有DataFrame
finalDF = reduce(DataFrame.union, dfs)
finalDF.show()

关键注意事项 #

• 所有读取的DataFrame必须和目标finalDF的schema完全一致（列名、顺序、数据类型均需匹配），否则union会报错。
• 读取CSV时，建议显式指定schema，避免Spark自动推断类型导致的不匹配。
• Parquet文件自带schema，若与目标schema不符，需要手动调整列和类型后再合并。

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