将 csv 文件与不匹配的列组合在一起答案

【问题标题】：Combining csv files with mismatched columns将 csv 文件与不匹配的列组合在一起
【发布时间】：2018-08-06 13:14:11
【问题描述】：

我需要将多个 csv 文件合并到一个对象中（我假设是一个数据框），但它们都有不匹配的列，如下所示：

CSV A

store_location_key | product_key | collector_key | trans_dt | sales | units | trans_key

CSV B

collector_key | trans_dt | store_location_key | product_key | sales | units | trans_key

CSV C

collector_key | trans_dt | store_location_key |product_key | sales | units | trans_id

除此之外，我还需要将这些与另外两个具有匹配列的 csv 文件相匹配：

位置 CSV

store_location_key | region | province | city | postal_code | banner | store_num

产品 CSV

product_key | sku | item_name | item_description | department | category

数据类型都是一致的，即sales列始终是float，store_location_key始终是int等。即使我先将每个csv转换为数据框，我也不确定join是否可以工作（除了最后两个）因为列需要匹配的方式。

【问题讨论】：

trans_key 和 trans_id 应该是相同的还是不同的列？
感谢您的回复...我不确定，但我认为最好将它们视为不同的列。
一两个文件的示例输入会很有帮助，也请分享您迄今为止尝试过的内容。

标签： csv apache-spark pyspark spark-dataframe data-analysis

【解决方案1】：

要合并前三个 CSV 文件，首先将其单独读取为 DataFrames，然后使用union。使用union 时的列顺序和数量很重要，因此首先您需要将所有缺失的列添加到DataFrame，然后使用select 确保列的顺序相同。

all_columns = ['collector_key', 'trans_dt', 'store_location_key', 'product_key', 'sales', 'units', 'trans_key', 'trans_id']

dfA = (spark.read.csv("a.csv", header=True)
  .withColumn(trans_id, lit(null))
  .select(all_columns))
dfB = (spark.read.csv("b.csv", header=True)
  .withColumn(trans_id, lit(null))
  .select(all_columns))
dfC = (spark.read.csv("c.csv", header=True)
  .withColumn(trans_key, lit(null))
  .select(all_columns))

df = dfA.union(dfB).union(dfC)

注意：如果 CSV 文件的列顺序/列数相同，则可以使用单个 spark.read 操作轻松组合它们。

合并前三个 CSV 后，剩下的就很简单了。位置和产品 CSV 都可以使用 join 与其余部分组合。

df_location = spark.read.csv("location.csv", header=True)
df_product = spark.read.csv("product.csv", header=True)

df2 = df.join(df_location, df.store_location_key == df_location.store_location_key)
df3 = df2.join(df_product, df2.product_key == df_product.product_key)

【讨论】：