如何在PySpark中读取多列不同的CSV并合并为单一DataFrame

文件结构说明 #

四个CSV的列结构如下：

capstone_customers.csv: [customer_id, customer_type, repeat_customer]
capstone_invoices.csv: [invoice_id, product_id, customer_id, days_until_shipped, product_line, total]
capstone_recent_customers.csv: [customer_id, customer_type]
capstone_recent_invoices.csv: [invoice_id, product_id, customer_id, days_until_shipped, product_line, total]

你的代码及报错 #

你尝试直接用union合并所有DataFrame，但因为列数、列结构不匹配报错：

你的代码 #

df1 = spark.read.options(inferSchema='True',header='True',delimiter=',').csv("capstone_customers.csv")
df2 = spark.read.options(inferSchema='True',header='True',delimiter=',').csv("capstone_invoices.csv")
df3 = spark.read.options(inferSchema='True',header='True',delimiter=',').csv("capstone_recent_customers.csv")
df4 = spark.read.options(inferSchema='True',header='True',delimiter=',').csv("capstone_recent_invoices.csv")

from functools import reduce
def unite_dfs(df1, df2):
  return df2.union(df1)

list_of_dfs = [df1, df2,df3,df4]
united_df = reduce(unite_dfs, list_of_dfs)

报错信息 #

Union只能在列数相同的表上执行，但第一个表有6列，第二个表有3列;;
'Union
:- Relation[invoice_id#234,product_id#235,customer_id#236,days_until_shipped#237,product_line#238,total#239] csv
+- Relation[customer_id#218,customer_type#219,repeat_customer#220] csv

解决步骤 #

1. 先合并同类型的数据集 #

union要求两个DataFrame的列数、列名、列顺序完全一致，所以先把客户表和发票表分别合并：

合并客户表

capstone_customers比recent版多了repeat_customer列，需要给recent表补上这一列（填充null）再合并：

from pyspark.sql.functions import lit

# 给df3添加repeat_customer列，匹配df1的列类型
df3_with_repeat = df3.withColumn("repeat_customer", lit(None).cast(df1.select("repeat_customer").schema[0].dataType))
# 合并两个客户表，用unionByName避免列顺序问题
combined_customers = df1.unionByName(df3_with_repeat)
# 按customer_id去重，保留唯一客户记录
combined_customers = combined_customers.dropDuplicates(["customer_id"])

合并发票表

两个发票表结构完全一致，直接用unionByName合并即可：

combined_invoices = df2.unionByName(df4)
# 按invoice_id去重，保留唯一发票记录
combined_invoices = combined_invoices.dropDuplicates(["invoice_id"])

2. 合并为单一DataFrame（关联客户与发票） #

如果需要把客户和发票数据整合到一个DF中，应该用join（按customer_id关联），这是业务逻辑上更合理的做法：

# 外连接保留所有客户和发票数据，即使没有匹配项
final_df = combined_customers.join(combined_invoices, on="customer_id", how="outer")

（可选）强行将所有行合并到一个DF（业务上不推荐） #

如果硬要把客户和发票的行放在同一个DF里，需要先让所有DF拥有全部列，缺失列填充null，再执行union：

from pyspark.sql.functions import lit

# 获取所有列的集合
all_columns = list(set(df1.columns + df2.columns + df3.columns + df4.columns))

# 定义函数：给DF补充缺失列并调整列顺序
def standardize_df(df, target_columns):
    for col in target_columns:
        if col not in df.columns:
            df = df.withColumn(col, lit(None))
    return df.select(target_columns)

# 标准化所有DataFrame
df1_full = standardize_df(df1, all_columns)
df2_full = standardize_df(df2, all_columns)
df3_full = standardize_df(df3, all_columns)
df4_full = standardize_df(df4, all_columns)

# 合并所有DF并去重
united_df = df1_full.union(df2_full).union(df3_full).union(df4_full)
united_df = united_df.dropDuplicates()

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