python - 如何使用另一个数据帧的新值更新 pyspark 数据帧？

Question

我有两个 Spark 数据框:

数据框A:

|col_1 | col_2 | ... | col_n |
|val_1 | val_2 | ... | val_n |

和数据框B:

|col_1 | col_2 | ... | col_m |
|val_1 | val_2 | ... | val_m |

数据帧 B 可以包含数据帧 A 中的重复行、更新行和新行。我想在 Spark 中编写一个操作，在其中可以创建一个新数据帧，其中包含数据帧 A 中的行以及数据帧 B 中更新的行和新行。

我首先创建一个仅包含不可更新列的哈希列。这是唯一的 ID。因此，假设 col1 和 col2 可以更改值(可以更新)，但 col3,..,coln 是唯一的。我创建了一个哈希函数 hash(col3,..,coln):

A=A.withColumn("hash", hash(*[col(colname) for colname in unique_cols_A]))
B=B.withColumn("hash", hash(*[col(colname) for colname in unique_cols_B]))

现在我想编写一些 Spark 代码，基本上从 B 中选择哈希值不在 A 中的行(因此新行和更新的行) 并将它们与来自 A 的行。如何在 pyspark 中实现此目的？

编辑:数据帧 B 可以包含数据帧 A 中的额外列，因此不可能进行并集。

示例示例

数据框A:

+-----+-----+
|col_1|col_2|
+-----+-----+
|    a|  www|
|    b|  eee|
|    c|  rrr|
+-----+-----+

数据框B:

+-----+-----+-----+
|col_1|col_2|col_3|
+-----+-----+-----+
|    a|  wew|    1|
|    d|  yyy|    2|
|    c|  rer|    3|
+-----+-----+-----+

结果:数据框C:

+-----+-----+-----+
|col_1|col_2|col_3|
+-----+-----+-----+
|    a|  wew|    1|
|    b|  eee| null|
|    c|  rer|    3|
|    d|  yyy|    2|
+-----+-----+-----+

Answer 1

这与 update a dataframe column with new values 密切相关，除了您还想添加 DataFrame B 中的行。一种方法是首先执行链接问题中概述的操作，然后将结果与 DataFrame B 合并并删除重复项。

例如:

dfA.alias('a').join(dfB.alias('b'), on=['col_1'], how='left')\
    .select(
        'col_1',
        f.when(
            ~f.isnull(f.col('b.col_2')),
            f.col('b.col_2')
        ).otherwise(f.col('a.col_2')).alias('col_2'),
        'b.col_3'
    )\
    .union(dfB)\
    .dropDuplicates()\
    .sort('col_1')\
    .show()
#+-----+-----+-----+
#|col_1|col_2|col_3|
#+-----+-----+-----+
#|    a|  wew|    1|
#|    b|  eee| null|
#|    c|  rer|    3|
#|    d|  yyy|    2|
#+-----+-----+-----+

或者更一般地使用列表理解，如果您有很多列需​​要替换并且您不想对它们进行硬编码:

cols_to_update = ['col_2']

dfA.alias('a').join(dfB.alias('b'), on=['col_1'], how='left')\
    .select(
        *[
            ['col_1'] + 
            [
                f.when(
                    ~f.isnull(f.col('b.{}'.format(c))),
                    f.col('b.{}'.format(c))
                ).otherwise(f.col('a.{}'.format(c))).alias(c)
                for c in cols_to_update
            ] + 
            ['b.col_3']
        ]
    )\
    .union(dfB)\
    .dropDuplicates()\
    .sort('col_1')\
    .show()