如何在PySpark读取时合并多个CSV文件并得到指定关联结果？

要实现按x列关联多个CSV文件、保留各文件对应列并将缺失值显示为NaN的需求，核心是用全外连接（Full Outer Join）替代行合并（Union），同时通过遍历目录实现批量读取文件。以下是具体实现步骤：

1. 基础实现（以两个文件为例） #

假设你的CSV文件包含表头（如first.csv有x,y列，second.csv有x,z列），先实现两个文件的关联：

from pyspark.sql import SparkSession
from pyspark.sql.functions import when, lit, isnull

# 初始化SparkSession
spark = SparkSession.builder.appName("CSVJoin").getOrCreate()

# 读取两个CSV文件（指定header=True表示文件有表头，inferSchema=True自动推断列类型）
df_first = spark.read.csv("first.csv", header=True, inferSchema=True)
df_second = spark.read.csv("second.csv", header=True, inferSchema=True)

# 按x列做全外连接，保留两边所有x值
joined_df = df_first.join(df_second, on="x", how="full_outer")

# 将Spark中的null替换为NaN（若转成Pandas DataFrame，null会自动转为NaN，此步骤可选）
for col_name in joined_df.columns:
    if col_name != "x":
        joined_df = joined_df.withColumn(
            col_name,
            when(isnull(col_name), lit(float("nan"))).otherwise(col_name)
        )

# 查看结果
joined_df.show()

执行后会得到包含x,y,z的结果集：

• 仅在first.csv存在的x值，对应z列显示NaN
• 仅在second.csv存在的x值，对应y列显示NaN
• 两边都存在的x值，显示完整的y和z值

2. 扩展支持读取目录下所有CSV文件 #

如果要批量处理目录下的所有CSV文件，可通过遍历目录逐个读取并依次做全外连接：

import os
from pyspark.sql import SparkSession
from pyspark.sql.functions import when, lit, isnull

spark = SparkSession.builder.appName("BatchCSVJoin").getOrCreate()

# 替换为你的CSV目录路径
csv_dir = "/path/to/your/csv/folder"

# 获取目录下所有CSV文件
csv_files = [f for f in os.listdir(csv_dir) if f.lower().endswith(".csv")]

# 初始化结果DataFrame
final_df = None

for file in csv_files:
    file_path = os.path.join(csv_dir, file)
    # 读取当前文件
    current_df = spark.read.csv(file_path, header=True, inferSchema=True)
    
    # 跳过没有x列的文件（根据需求可调整处理逻辑）
    if "x" not in current_df.columns:
        print(f"跳过文件 {file}：未找到x列")
        continue
    
    # 处理列名冲突（可选：若不同文件有相同非x列名，给列名添加文件名前缀）
    file_prefix = os.path.splitext(file)[0]
    for col in current_df.columns:
        if col != "x":
            current_df = current_df.withColumnRenamed(col, f"{col}_{file_prefix}")
    
    # 第一次读取时直接赋值，后续做全外连接
    if final_df is None:
        final_df = current_df
    else:
        final_df = final_df.join(current_df, on="x", how="full_outer")

# 将null替换为NaN
if final_df is not None:
    for col_name in final_df.columns:
        if col_name != "x":
            final_df = final_df.withColumn(
                col_name,
                when(isnull(col_name), lit(float("nan"))).otherwise(col_name)
            )
    final_df.show()
else:
    print("目录下无符合要求的CSV文件")

关键注意事项 #

• 表头与Schema：确保CSV文件包含表头，若没有表头需手动指定schema参数，避免Spark默认生成_c0这类列名。
• 列名冲突处理：如果不同文件存在相同的非x列名，必须通过重命名（如添加文件名前缀）避免join时的列名冲突。
• 缺失值显示：Spark原生用null表示缺失值，若需在Spark控制台显示为NaN，需手动替换；若将结果转为Pandas DataFrame，null会自动转为NaN。

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