假设我们有两个数据帧,我们想通过左反连接来比较它们的差异:
data1 = [
(1, 11, 20, None),
(2, 12, 22, 31),
]
data2 = [
(1, 11, 20, None),
(2, 12, 22, 31),
]
schema = StructType([ \
StructField("value_1",IntegerType(), True), \
StructField("value_2",IntegerType(), True), \
StructField("value_3",IntegerType(), True), \
StructField("value_4",IntegerType(), True), \
])
df1 = spark.createDataFrame(data=data1,schema=schema)
df2 = spark.createDataFrame(data=data2,schema=schema)
如何通过多个(所有)列 nullsafe 连接这些数据帧? 我想出的唯一解决方案如下:
df = df1.join(df2, \
((df1.value_1.eqNullSafe(df2.value_1)) &
(df1.value_2.eqNullSafe(df2.value_2)) &
(df1.value_3.eqNullSafe(df2.value_3)) &
(df1.value_4.eqNullSafe(df2.value_4))),
"leftanti" \
)
但不幸的是,我们现在必须处理大量列的动态列表。 我们如何以某种方式重写此连接,以便我们可以提供要连接的列的列表。
感谢和BR
最佳答案
据我理解问题陈述,您希望根据提供的列列表创建动态连接条件。我们可以使用 functools
模块中的 reduce()
来做到这一点。
join_cols = ['value_1', 'value_2', 'value_3', 'value_4']
from functools import reduce
join_condition = reduce(lambda x, y: x & y, [df1[k].eqNullSafe(df2[k]) for k in join_cols])
print(join_condition)
# Column<'((((value_1 <=> value_1) AND (value_2 <=> value_2)) AND (value_3 <=> value_3)) AND (value_4 <=> value_4))'>
您可以直接在.join()
中使用join_condition
参数。
df = df1.join(df2, join_condition, "leftanti")
关于python - PYSPARK - 在多个列上加入 nullsafe,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/72799538/