我如何读取多个 csv 文件并将它们合并到 pyspark 中的单个数据框中答案

【问题标题】：how can i read multiple csv files and merge them in single dataframe in pyspark我如何读取多个 csv 文件并将它们合并到 pyspark 中的单个数据框中
【发布时间】：2022-10-12 22:08:56
【问题描述】：

我有 4 个不同列的 CSV 文件。一些 csv 也具有相同的列名。 csv的详细信息是：

capstone_customers.csv：[customer_id，customer_type，repeat_customer]

capstone_invoices.csv：[invoice_id、product_id、customer_id、days_until_shipped、product_line、总计]

capstone_recent_customers.csv：[customer_id，customer_type]

capstone_recent_invoices.csv：[invoice_id、product_id、customer_id、days_until_shipped、product_line、total]

我的代码是

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列；\n'Union\n:- Relation[invoice_id#234,product_id#235,customer_id #236,days_until_shipped#237,product_line#238,total#239] csv\n+- 关系[customer_id#218,customer_type#219,repeat_customer#220] csv\n"

如何在单个数据框中合并并使用 pyspark 删除相同的列名

【问题讨论】：

标签： python csv apache-spark pyspark

【解决方案1】：

您可以提供要读取的文件列表或文件路径，而不是一一读取。不要忘记mergeSchema 选项：

files = [
   "capstone_customers.csv",
   "capstone_invoices.csv",
   "capstone_recent_customers.csv",
   "capstone_recent_invoices.csv"
]
df = spark.read.options(inferSchema='True',header='True',delimiter=',', mergeSchema='True').csv(files)

# or
df = spark.read.options(inferSchema='True',header='True',delimiter=',',mergeSchema='True').csv('/path/to/files/')

【讨论】：