如何基于组加入 Pyspark 数据帧答案

【问题标题】：How to join Pyspark dataframes based on groups如何基于组加入 Pyspark 数据帧
【发布时间】：2021-09-29 04:39:16
【问题描述】：

我有 2 个 Pyspark 数据帧

数据框 1 - df 列是 customer_id, address_id, order_id, date the order was placed, order_share

+----+----+--------+----------+-----------+
|c_id|a_id|order_id|order_date|order_share|
+----+----+--------+----------+-----------+
|  c1|  a1|       1|2021-01-23|        0.5|
|  c1|  a2|       1|2021-01-23|        0.2|
|  c1|  a3|       1|2021-01-23|        0.3|
|  c2|  a5|       2|2021-03-20|        0.4|
|  c2|  a6|       2|2021-03-20|        0.6|
|  c1|  a1|       3|2021-02-20|        0.3|
|  c1|  a2|       3|2021-02-20|        0.3|
|  c1|  a3|       3|2021-02-20|        0.4|
+----+----+--------+----------+-----------+

数据框 2 - df_address 列是 customer_id, address_id, the date of address creation

+----+----+------------+
|c_id|a_id|created_date|
+----+----+------------+
|  c1|  a1|  2020-12-31|
|  c1|  a2|  2020-04-23|
|  c1|  a3|  2020-03-23|
|  c1|  a4|  2020-01-16|
|  c2|  a5|  2020-12-28|
|  c2|  a6|  2020-05-16|
|  c2|  a7|  2020-03-04|
+----+----+------------+

现在，我希望加入这两个表，以便对于 每个 order_id，我从 df_address 获取地址，并且对应的条目应该是 order_share 列中的 0.0 p>

我的输出应该是这样的

+----+----+------------+--------+----------+-----------+
|c_id|a_id|created_date|order_id|order_date|order_share|
+----+----+------------+--------+----------+-----------+
|  c1|  a1|  2020-12-31|       1|2021-01-23|        0.5|
|  c1|  a2|  2020-04-23|       1|2021-01-23|        0.2|
|  c1|  a3|  2020-03-23|       1|2021-01-23|        0.3|
|  c1|  a4|  2020-01-16|       1|2021-01-23|        0.0|
|  c2|  a5|  2020-12-28|       2|2021-03-20|        0.4|
|  c2|  a6|  2020-05-16|       2|2021-03-20|        0.6|
|  c2|  a7|  2020-03-04|       2|2021-03-20|        0.0|
|  c1|  a1|  2020-12-31|       3|2021-02-20|        0.3|
|  c1|  a2|  2020-04-23|       3|2021-02-20|        0.3|
|  c1|  a3|  2020-03-23|       3|2021-02-20|        0.4|
|  c1|  a4|  2020-01-16|       3|2021-02-20|        0.0|
+----+----+------------+--------+----------+-----------+

这看起来不像普通的左/右连接，我应该为每个 order_id 都这样做。

我尝试使用['c_id','a_id'] 加入，但输出与预期相差甚远。考虑到df_address 左边和df 右边，使用left join 给我order_id 和right join 的空值并没有给我来自df_address 的所有地址

看起来我必须为每个 order_id 应用某种 groupby，然后为每个组应用 join，但我不知道如何实现，甚至不知道这是否是正确的方法

任何帮助将不胜感激。谢谢！

【问题讨论】：

什么阻止你加入left c_id 和 a_id
请检查问题的编辑，我已经说明了我在使用左连接时遇到的问题

标签： python apache-spark join pyspark

【解决方案1】：

您可以使用中间的orders 数据框，它是从df 数据框创建的，并且仅包含有关订单的信息，即列customer_id、order_id 和order_date。然后，您首先将 df_address 数据帧与此 orders 数据帧进行内部连接，将每对 (customer_id, address_id) 链接到特定于订单的信息，然后将生成的数据帧与 df 数据帧左连接以获得每个地址的order_share，然后将order_share 列中的null 值替换为0.0。

完整代码如下：

from pyspark.sql import functions as F

# Orders dataframe that contains only orders-specific information
orders = df.select('customer_id', 'order_id', 'order_date').distinct()

df_address.join(orders, ['customer_id']) \ # link addresses with orders
  .join(df.drop('order_date'), ['customer_id', 'address_id', 'order_id'], 'left_outer') \ # link orders/addresses with order shares
  .withColumn('order_share', F.when(F.col('order_share').isNotNull(), F.col('order_share')).otherwise(F.lit(0.0))) \ # replace null in order_share column with 0.0
  .orderBy('customer_id', 'order_id', 'address_id') \ # optional, to reorder dataframe

详情

注意：为了便于阅读，我在这里按 order_id 和 address_id 对所有数据帧进行了重新排序

从您问题中的df 数据框开始，我们得到以下orders 数据框：

+-----------+--------+----------+
|customer_id|order_id|order_date|
+-----------+--------+----------+
|c1         |1       |2021-01-23|
|c2         |2       |2021-03-20|
|c1         |3       |2021-02-20|
+-----------+--------+----------+

然后我们将这个orders 数据框与df_address 数据框连接起来：

+-----------+----------+------------+--------+----------+
|customer_id|address_id|created_date|order_id|order_date|
+-----------+----------+------------+--------+----------+
|c1         |a1        |2020-12-31  |1       |2021-01-23|
|c1         |a2        |2020-04-23  |1       |2021-01-23|
|c1         |a3        |2020-03-23  |1       |2021-01-23|
|c1         |a4        |2020-01-16  |1       |2021-01-23|
|c2         |a5        |2020-12-28  |2       |2021-03-20|
|c2         |a6        |2020-05-16  |2       |2021-03-20|
|c2         |a7        |2020-03-04  |2       |2021-03-20|
|c1         |a1        |2020-12-31  |3       |2021-02-20|
|c1         |a2        |2020-04-23  |3       |2021-02-20|
|c1         |a3        |2020-03-23  |3       |2021-02-20|
|c1         |a4        |2020-01-16  |3       |2021-02-20|
+-----------+----------+------------+--------+----------+

最后一次加入 df 数据框，没有列 order_date，我们得到：

+-----------+----------+--------+------------+----------+-----------+
|customer_id|address_id|order_id|created_date|order_date|order_share|
+-----------+----------+--------+------------+----------+-----------+
|c1         |a1        |1       |2020-12-31  |2021-01-23|0.5        |
|c1         |a2        |1       |2020-04-23  |2021-01-23|0.2        |
|c1         |a3        |1       |2020-03-23  |2021-01-23|0.3        |
|c1         |a4        |1       |2020-01-16  |2021-01-23|null       |
|c2         |a5        |2       |2020-12-28  |2021-03-20|0.4        |
|c2         |a6        |2       |2020-05-16  |2021-03-20|0.6        |
|c2         |a7        |2       |2020-03-04  |2021-03-20|null       |
|c1         |a1        |3       |2020-12-31  |2021-02-20|0.3        |
|c1         |a2        |3       |2020-04-23  |2021-02-20|0.3        |
|c1         |a3        |3       |2020-03-23  |2021-02-20|0.4        |
|c1         |a4        |3       |2020-01-16  |2021-02-20|null       |
+-----------+----------+--------+------------+----------+-----------+

然后我们只需将null 替换为0.0 即可得到预期的数据帧：

+-----------+----------+--------+------------+----------+-----------+
|customer_id|address_id|order_id|created_date|order_date|order_share|
+-----------+----------+--------+------------+----------+-----------+
|         c1|        a1|       1|  2020-12-31|2021-01-23|        0.5|
|         c1|        a2|       1|  2020-04-23|2021-01-23|        0.2|
|         c1|        a3|       1|  2020-03-23|2021-01-23|        0.3|
|         c1|        a4|       1|  2020-01-16|2021-01-23|        0.0|
|         c2|        a5|       2|  2020-12-28|2021-03-20|        0.4|
|         c2|        a6|       2|  2020-05-16|2021-03-20|        0.6|
|         c2|        a7|       2|  2020-03-04|2021-03-20|        0.0|
|         c1|        a1|       3|  2020-12-31|2021-02-20|        0.3|
|         c1|        a2|       3|  2020-04-23|2021-02-20|        0.3|
|         c1|        a3|       3|  2020-03-23|2021-02-20|        0.4|
|         c1|        a4|       3|  2020-01-16|2021-02-20|        0.0|
+-----------+----------+--------+------------+----------+-----------+

【讨论】：

谢谢。只是一些澄清，我不认为 distinct 在创建 orders 时真的需要，因为该行总是独一无二的？而且，在加入时使用left_outer 或left 意味着同样的事情对吧？
关于无用的distinct，只有customer_id、order_id 和order_date 的行不会是唯一的，因为我们删除了address_id 和order_share 列。以您的示例df 数据框为例，如果您删除distinct，则在orders 数据框中您将获得三倍(c1, 1, 2021-01-23) 行、两次(c2, 2, 2021-03-20) 行和三倍(c1, 3, 2021-02-20) 行。所以你必须保留distinct。是的，left_outer 和 left 在连接两个数据帧时的含义相同。
对不起，我的错。我有别的想法。我了解distinct 的用法。我只是澄清一下，我们将distinct 与dropDuplicates 互换使用，对吗？
是的，您可以在这里互换使用 distinct 和 dropDuplicates 而没有参数。

【解决方案2】：

我尝试将full outer 与DataFrames 连接起来，以获取缺少的c_id 和a_id 组合，并进一步利用when 和isNull 来获得@987654325来自df 的@ 列值，并在下面替换来自df_address 的值是结果-

数据准备

input_str1 = """
c1|  a1|       1|2021-01-23|        0.5|
c1|  a2|       1|2021-01-23|        0.2|
c1|  a3|       1|2021-01-23|        0.3|
c2|  a5|       2|2021-03-20|        0.4|
c2|  a6|       2|2021-03-20|        0.6|
c1|  a1|       3|2021-02-20|        0.3|
c1|  a2|       3|2021-02-20|        0.3|
c1|  a3|       3|2021-02-20|        0.4
""".split("|")

input_values1 = list(map(lambda x: x.strip() if x.strip() != '' else None, input_str1))

cols1 = list(map(lambda x: x.strip() if x.strip() != '' else None, "c_id|a_id|order_id|order_date|order_share".split("|")))
            
n = len(input_values1)
n_col1 = 5

input_list1 = [tuple(input_values1[i:i+n_col1]) for i in range(0,n,n_col1)]

sparkDF1 = sql.createDataFrame(input_list1, cols1)

sparkDF1.show()

+----+----+--------+----------+-----------+
|c_id|a_id|order_id|order_date|order_share|
+----+----+--------+----------+-----------+
|  c1|  a1|       1|2021-01-23|        0.5|
|  c1|  a2|       1|2021-01-23|        0.2|
|  c1|  a3|       1|2021-01-23|        0.3|
|  c2|  a5|       2|2021-03-20|        0.4|
|  c2|  a6|       2|2021-03-20|        0.6|
|  c1|  a1|       3|2021-02-20|        0.3|
|  c1|  a2|       3|2021-02-20|        0.3|
|  c1|  a3|       3|2021-02-20|        0.4|
+----+----+--------+----------+-----------+

input_str2 = """
c1|  a1|  2020-12-31|
c1|  a2|  2020-04-23|
c1|  a3|  2020-03-23|
c1|  a4|  2020-01-16|
c2|  a5|  2020-12-28|
c2|  a6|  2020-05-16|
c2|  a7|  2020-03-04
""".split("|")

input_values2 = list(map(lambda x: x.strip() if x.strip() != '' else None, input_str2))

cols2 = list(map(lambda x: x.strip() if x.strip() != '' else None, "c_id|a_id|created_date".split("|")))
            
n = len(input_values2)
n_col2 = 3

input_list2 = [tuple(input_values2[i:i+n_col2]) for i in range(0,n,n_col2)]

sparkDF2 = sql.createDataFrame(input_list2, cols2)

sparkDF2.show()

+----+----+------------+
|c_id|a_id|created_date|
+----+----+------------+
|  c1|  a1|  2020-12-31|
|  c1|  a2|  2020-04-23|
|  c1|  a3|  2020-03-23|
|  c1|  a4|  2020-01-16|
|  c2|  a5|  2020-12-28|
|  c2|  a6|  2020-05-16|
|  c2|  a7|  2020-03-04|
+----+----+------------+

完全加入

重命名来自 SparkDF2 的列值，这将进一步用于填充空值以避免列名不明确

finalDF = sparkDF1.join(sparkDF2
                       , (sparkDF1['c_id'] == sparkDF2['c_id'])
                        & (sparkDF1['a_id'] == sparkDF2['a_id'])
                        ,'full'
                ).select(sparkDF1['*']
                         ,sparkDF2['c_id'].alias('c_id_address')
                         ,sparkDF2['a_id'].alias('a_id_address')
                         ,sparkDF2['created_date']
                        )
finalDF.show()

+----+----+--------+----------+-----------+------------+------------+------------+
|c_id|a_id|order_id|order_date|order_share|c_id_address|a_id_address|created_date|
+----+----+--------+----------+-----------+------------+------------+------------+
|  c1|  a3|       1|2021-01-23|        0.3|          c1|          a3|  2020-03-23|
|  c1|  a3|       3|2021-02-20|        0.4|          c1|          a3|  2020-03-23|
|  c2|  a5|       2|2021-03-20|        0.4|          c2|          a5|  2020-12-28|
|null|null|    null|      null|       null|          c2|          a7|  2020-03-04|
|  c1|  a2|       1|2021-01-23|        0.2|          c1|          a2|  2020-04-23|
|  c1|  a2|       3|2021-02-20|        0.3|          c1|          a2|  2020-04-23|
|  c1|  a1|       1|2021-01-23|        0.5|          c1|          a1|  2020-12-31|
|  c1|  a1|       3|2021-02-20|        0.3|          c1|          a1|  2020-12-31|
|null|null|    null|      null|       null|          c1|          a4|  2020-01-16|
|  c2|  a6|       2|2021-03-20|        0.6|          c2|          a6|  2020-05-16|
+----+----+--------+----------+-----------+------------+------------+------------+

当为空时

finalDF = finalDF.withColumn('c_id',F.when(F.col('c_id').isNull()
                                           ,F.col('c_id_address')).otherwise(F.col('c_id'))
                            )\
                    .withColumn('a_id',F.when(F.col('a_id').isNull()
                                              ,F.col('a_id_address')).otherwise(F.col('a_id'))
                            )\
                    .withColumn('order_share',F.when(F.col('order_share').isNull()
                                                     ,0.0).otherwise(F.col('order_share'))
                            )


finalDF.show()

+----+----+--------+----------+-----------+------------+------------+------------+
|c_id|a_id|order_id|order_date|order_share|c_id_address|a_id_address|created_date|
+----+----+--------+----------+-----------+------------+------------+------------+
|  c1|  a3|       1|2021-01-23|        0.3|          c1|          a3|  2020-03-23|
|  c1|  a3|       3|2021-02-20|        0.4|          c1|          a3|  2020-03-23|
|  c2|  a5|       2|2021-03-20|        0.4|          c2|          a5|  2020-12-28|
|  c2|  a7|    null|      null|        0.0|          c2|          a7|  2020-03-04|
|  c1|  a2|       1|2021-01-23|        0.2|          c1|          a2|  2020-04-23|
|  c1|  a2|       3|2021-02-20|        0.3|          c1|          a2|  2020-04-23|
|  c1|  a1|       1|2021-01-23|        0.5|          c1|          a1|  2020-12-31|
|  c1|  a1|       3|2021-02-20|        0.3|          c1|          a1|  2020-12-31|
|  c1|  a4|    null|      null|        0.0|          c1|          a4|  2020-01-16|
|  c2|  a6|       2|2021-03-20|        0.6|          c2|          a6|  2020-05-16|
+----+----+--------+----------+-----------+------------+------------+------------+

注意 - order_id 和 order_date 为空，因为 sparkDF2 中的 c_id 和 a_id 组合不存在值

此示例提供了一种方法，以获得所需的解决方案，如果需要填充订单缺失值，您可以进一步即兴发挥

【讨论】：