当列数未知时，如何在多个列上连接两个表（pyspark）

Posted 2023-04-15

【中文标题】当列数未知时，如何在多个列上连接两个表（pyspark）

【英文标题】：How do I join two tables on multiple columns when number of columns is unknown (pyspark)

【发布时间】：2020-09-02 13:53:58

【问题描述】：

假设我有两个表 A 和 B。让它们的结构类似于：

A:

------------------------------
col_1 | col_2 | col_3 | col_4
------------------------------
   1  |  A    |  a    |  i
   2  |  B    |  b    |  ii
   3  |  C    |  c    |  iii
   4  |  D    |  d    |  iv
   5  |  E    |  e    |  v
------------------------------

B:

---------------
col_1 | col_3 
---------------
  1   |  null
  3   |  c
  null|  b
  2   |  null
--------------

确定 B 中的列名与 A 中的相同，我想在各个列中使用 OR 条件加入它们。唯一的问题是 B 中的列数是未知的。

如何执行连接？

我想做的伪代码如下：

select *
from A
join B
on A.col_1 == B.col_1
OR A.col_2 == B.col_2
......
OR A.col_k == B.col_k   --where k is the total number of columns in B

我为spark.sql 创建了以下字符串，但我正在寻找一种更Pyspark-ic 的方式：

sql_query = 'select s.* from dfA s join dfB on '

#join using or conditions
for i in dfB.columns:
    sql_query += 'dfA.' +i + ' == dfB.' + i + ' OR '

#remove the last extra 'OR'
sql_query = sql_query[:-3]

spark.sql(sql_query)

上述方法需要创建临时视图，以便可以在sqlContext 中访问它们。

【问题讨论】：

一张表的列数怎么不知道？在任何情况下，SQL 作为一般规则在join 条件中对可变数量的列没有任何语法。

B 正在使用某些逻辑自动生成。如果可能，我想遍历 B 中的所有列。以上已在 Pyspark 中实现。

我的意思是，B中的列数只能在运行时确定。

【参考方案1】：

Dataframe.columns 返回数据框的列列表。通过这个属性，我们可以获得两个数据框共有的列：

dfA = ...
dfB = ...

#get the common columns
common_cols = [col for col in dfA.columns if col in dfB.columns]

#create a list of join conditions
join_conds = [dfA[col].eqNullSafe(dfB[col]) for col in common_cols]

#combine all join conditions with "or"
cond = join_conds[0]
for c in join_conds[1:]:
    cond = cond | c

#use the combined condition in a join
dfA.join(dfB, cond).show()

【讨论】：

@ghost 。 . .数据框方法可能是最好的解决方案。

@werner，我添加了我遵循的方法。这正是我一直在寻找的答案。