Pyspark - 将具有不同架构的文件合并到一个主文件中

Question

我有九个如下所示的 csv 文件：

trans_1
+------------------+-----------+-------------+----------+-----+-----+--------------------+
|store_location_key|product_key|collector_key|  trans_dt|sales|units|           trans_key|
+------------------+-----------+-------------+----------+-----+-----+--------------------+
|              9807|83215400105|           -1|2015-09-09|42.72|    1|19815980756712015...|
|              9807| 6024538816|           -1|2015-10-28|27.57|    1|21718980756712015...|
+------------------+-----------+-------------+----------+-----+-----+--------------------+
only showing top 2 rows

trans_2
+------------------+-----------+-------------+----------+-----+-----+--------------------+
|store_location_key|product_key|collector_key|  trans_dt|sales|units|           trans_key|
+------------------+-----------+-------------+----------+-----+-----+--------------------+
|              7296|85375900278|           -1|2015-06-26| 4.97|    1|12548729658922015...|
|              7296|81526001001| 139537965459|2015-05-01|44.48|    1|24990729650922015...|
+------------------+-----------+-------------+----------+-----+-----+--------------------+
only showing top 2 rows

trans_3
+------------------+-----------+-------------+----------+-----+-----+--------------------+
|store_location_key|product_key|collector_key|  trans_dt|sales|units|           trans_key|
+------------------+-----------+-------------+----------+-----+-----+--------------------+
|              9807|83215400105|           -1|2015-09-09|42.72|    1|19815980756712015...|
|              9807| 6024538816|           -1|2015-10-28|27.57|    1|21718980756712015...|
+------------------+-----------+-------------+----------+-----+-----+--------------------+
only showing top 2 rows

trans_4
+-------------+----------+------------------+-----------+-----+-----+----------------+
|collector_key|  trans_dt|store_location_key|product_key|sales|units|       trans_key|
+-------------+----------+------------------+-----------+-----+-----+----------------+
|           -1| 6/26/2015|              8142| 4319416816| 9.42|    1|1.6945500000E+25|
|           -1|10/25/2015|              8142| 6210700491| 24.9|    1|3.4001800000E+25|
+-------------+----------+------------------+-----------+-----+-----+----------------+
only showing top 2 rows

trans_5
+-------------+----------+------------------+---------------+-----+-----+--------------------+
|collector_key|  trans_dt|store_location_key|    product_key|sales|units|           trans_key|
+-------------+----------+------------------+---------------+-----+-----+--------------------+
|           -1|2015-10-28|              6973|999999999999513|  0.0|    1|31575569731182201...|
|           -1|2015-07-24|              6973|    77105810883| 8.53|    1|31216969731182201...|
+-------------+----------+------------------+---------------+-----+-----+--------------------+
only showing top 2 rows

trans_6
+-------------+----------+------------------+----------------+-----+-----+----------------+
|collector_key|  trans_dt|store_location_key|     product_key|sales|units|        trans_id|
+-------------+----------+------------------+----------------+-----+-----+----------------+
|           -1|10/28/2015|              6973|1000000000000000|  0.0| null|3.1575600000E+25|
|           -1| 7/24/2015|              6973|     77105810883| 8.53| null|3.1217000000E+25|
+-------------+----------+------------------+----------------+-----+-----+----------------+
only showing top 2 rows

trans_7
+-------------+----------+------------------+-----------+-----+-----+--------------------+
|collector_key|  trans_dt|store_location_key|product_key|sales|units|            trans_id|
+-------------+----------+------------------+-----------+-----+-----+--------------------+
|           -1|2015-09-09|              9807|83215400105|42.72|    1|19815980756712015...|
|           -1|2015-10-28|              9807| 6024538816|27.57|    1|21718980756712015...|
+-------------+----------+------------------+-----------+-----+-----+--------------------+
only showing top 2 rows

trans_8
+----------------+-------------+----------+------------------+-----+-----+----------------+
|     product_key|collector_key|  trans_dt|store_location_key|sales|units|        trans_id|
+----------------+-------------+----------+------------------+-----+-----+----------------+
|1000000000000000|           -1|10/28/2015|              6973| null|    1|3.1575600000E+25|
|     77105810883|           -1| 7/24/2015|              6973| null|    1|3.1217000000E+25|
+----------------+-------------+----------+------------------+-----+-----+----------------+
only showing top 2 rows

trans_9
+-----------+-------------+----------+------------------+-----+-----+--------------------+
|product_key|collector_key|  trans_dt|store_location_key|sales|units|            trans_id|
+-----------+-------------+----------+------------------+-----+-----+--------------------+
| 4319416816|           -1|2015-06-26|              8142| 9.42|    1|16945481425160201...|
| 6210700491|           -1|2015-10-25|              8142| 24.9|    1|34001814221225201...|
+-----------+-------------+----------+------------------+-----+-----+--------------------+
only showing top 2 rows

它们都具有相同的列但位于不同的位置。我使用此代码读取所有文件但出现错误。

trans = spark\
    .read\
    .format("csv")\
    .option("inferSchema","true")\
    .option("header","true")\
    .load("/Users/xyz/Downloads/xyz/trans_fact*.csv")

我只想在 pyspark 中编写代码，以便我可以读取所有这些文件并将它们合并到一个数据帧 (csv) 中，并在正确的顺序列下使用正确的数据。

Answer 1

您可以一一加载 csv 文件，添加可能缺少的列，对列进行排序，然后合并它们：

import os
def load_single_files(dir):    
    dirpath,_,files = next(os.walk(dir))
    for f in files:
        yield spark\
            .read\
            .format("csv")\
            .option("inferSchema","true")\
            .option("header","true")\
            .load(os.path.join(dirpath, f))

import collections
def add_missing_cols_in_order(df,unique_cols):
    missing_cols = {col:F.lit(None).alias(col) for col in unique_cols if col not in df.columns}
    existing_cols = {col:F.col(col) for col in unique_cols if col in df.columns}
    cols = dict(missing_cols, **existing_cols)
    cols = list(collections.OrderedDict(sorted(cols.items())).values())
    return df.select(cols)

dfs = list(load_single_files("testdata"))
unique_cols = sorted(set([col for cols in [df.columns for df in dfs] for col in cols]))
df = dfs[0]
df = add_missing_cols_in_order(df, unique_cols)

for next_df in dfs[1:]:
    df = df.union(add_missing_cols_in_order(next_df, unique_cols))

这种方法比一次加载所有文件要慢，因为 Spark 不会并行读取文件。根据文件的大小，这可能是也可能不是问题。

编辑：遵循thebluephantom's 的建议，包括自动添加缺失列的逻辑。