pyspark中的动态布尔连接答案

【问题标题】：Dynamic boolean join in pysparkpyspark中的动态布尔连接
【发布时间】：2019-12-13 05:02:50
【问题描述】：

我有两个具有相同架构的 pyspark 数据框，如下所示 -

df_source：

id, name, age

df_target：

id,name,age

“id”是表中的主列，其余是属性列

我接受用户的主要和属性列列表如下-

primary_columns = ["id"]
attribute_columns = ["name","age"]

我需要如下动态加入以上两个数据框 -

df_update = df_source.join(df_target, (df_source["id"] == df_target["id"]) & ((df_source["name"] != df_target["name"]) | (df_source["age"] != df_target["age"]))  ,how="inner").select([df_source[col] for col in df_source.columns])

我怎样才能在 pyspark 中动态地实现这个连接条件，因为属性和主键列的数量可以根据用户输入而改变？请帮忙。

【问题讨论】：

标签： pyspark

【解决方案1】：

IIUC，您只需使用primary_columns 上的内连接和循环attribute_columns 的where 子句即可实现所需的输出。

由于两个 DataFrame 具有相同的列名，因此在 join 之后使用 alias 来区分列名。

from functools import reduce
from pyspark.sql.functions import col

df_update = df_source.alias("s")\
    .join(df_target.alias("t"), on=primary_columns, how="inner")\
    .where(
         reduce(
             lambda a, b: a|b, 
             [(col("s."+c) != col("t."+c) for c in attribute_columns]
         )\
    )
    .select("s.*")

使用 reduce 对attribute_columns 中的列应用按位或运算。

【讨论】：