Spark数据框左连接应在右侧添加默认行而不是null的连接

Posted 2023-04-17

【中文标题】Spark数据框左连接应在右侧添加默认行而不是null的连接

【英文标题】：Spark dataframe left join should add join with default row instead of null on right side

【发布时间】：2019-07-30 18:37:28

【问题描述】：

我试图将 Pyspark 中的两个数据框留在一个公共列上。如果公共列的值不存在于右侧数据框中，则插入空值。我希望它与正确数据框中的默认行连接，而不是空值。有没有可能做到这一点？

df1 = df1.join(df2, [df1.id == df2.channel_id], "left")
df1.select('channel_id', 'channel_name').show()

我的代码为不匹配的行打印 null、null。相反，我希望它与 id = 0 的行匹配，这是默认值。

【问题讨论】：

所以本质上你想用id=0 的默认值替换Null 值？这可能比另一个连接更容易实现

这里需要设置多少列？

@NeilZ 可以有任意数量的列。我在这里想要的是，而不是右数据框中所有列的 Null 值，我想要右数据框中默认行的值。在我的情况下，我有 id = 0 的默认列。

添加一个minimal reproducible example，其中包含一小部分数据样本和所需的输出。

【参考方案1】：

这可能是实现您想要的一种方式。

(1) 首先让我们创建 2 个示例 DataFrames，然后将left 加入它们以获得df_join。

df1 = spark.createDataFrame([(1,'a'), (2,'b'),(3, 'f'),(4,'c'),(5, 'd')], ['id', 'name1'])
df2 = spark.createDataFrame([(1,'aaa'), (2,'bb'),(7, 'ff'),(4,'cc'),(8,'mm'), (10,'kk')], ['channel_id', 'channel_name'])
df1.show()
df2.show()
df_join = df1.join(df2, [df1.id == df2.channel_id], "left")
df_join.orderBy('id').show()

DataFrames 如下所示。您可以看到df2 中不存在id 3 和5 的行，因此它们在列中的对应值：channel_id 和channel_name 是null。我们希望用id==1 行中的相应值填充它们。

------ df1 ------
+---+-----+
| id|name1|
+---+-----+
|  1|    a|
|  2|    b|
|  3|    f|
|  4|    c|
|  5|    d|
+---+-----+

------ df2 ------
+----------+------------+
|channel_id|channel_name|
+----------+------------+
|         1|         aaa|
|         2|          bb|
|         7|          ff|
|         4|          cc|
|         8|          mm|
|        10|          kk|
+----------+------------+

------ df_join ------
+---+-----+----------+------------+
| id|name1|channel_id|channel_name|
+---+-----+----------+------------+
|  1|    a|         1|         aaa|
|  2|    b|         2|          bb|
|  3|    f|      null|        null|
|  4|    c|         4|          cc|
|  5|    d|      null|        null|
+---+-----+----------+------------+

(2) 接下来，选择你想要依赖的默认行，这里我只是使用了id==1 的行。

default_row = df_join.where(df_join['id']==1).select('channel_id', 'channel_name').take(1)[0].asDict()
print('----- default_row:  ------'.format(default_row))

输出显示我们将使用的默认值：

----- default_row: 'channel_id': 1, 'channel_name': 'aaa' ------

(3) 现在我们可以在每一列上使用fillna 填充所有空值

for k, v in default_row.items():
    df_join = df_join.fillna(v, subset=k)
df_join.orderBy('id').show()

最终的 DataFrame 如下所示。空值已成功填充。

+---+-----+----------+------------+
| id|name1|channel_id|channel_name|
+---+-----+----------+------------+
|  1|    a|         1|         aaa|
|  2|    b|         2|          bb|
|  3|    f|         1|         aaa|
|  4|    c|         4|          cc|
|  5|    d|         1|         aaa|
+---+-----+----------+------------+