基于另一个数据框的带有列条件的pyspark

Question

我有输入：

数据框 A：

valueX valueY
     1      2
     4      3
    -1      4

数据框 B：

startX endX startY endY
     0    2      2    4
     3    5     -2   -1
    -2    0      0    4

我想获得输出，将isIn 列添加到 A：

valueX valueY isIn
     1      2 True
     4      3 False 
    -1      4 True

条件 startX

我试试这个：

dataframeA.join(dataframeB).withColumn("isIn",sf.when((dataframeA["valueX"] > dataframeB["startX"]) & (dataframeA["valueX"] < dataframeB["endX"]) & (dataframeA["valueY"] > dataframeB["startY"]) & (dataframeA["valueY"] < dataframeB["endY"]),
                        True).otherwise(False)))

但我必须删除重复的 isIn 和 valueX、valueY 行。

Answer 1

您可以进行半连接和反连接并将结果合并：

import pyspark.sql.functions as F

dfa_joined = dfa.alias('a').join(
    dfb.alias('b'), 
    F.col('a.valueX').between(F.col('b.startX'), F.col('b.endX')) & 
    F.col('a.valueY').between(F.col('b.startY'), F.col('b.endY')), 
    'left_semi'
).withColumn('isIn', F.lit(True))

dfb_joined = dfa.alias('a').join(
    dfb.alias('b'), 
    F.col('a.valueX').between(F.col('b.startX'), F.col('b.endX')) & 
    F.col('a.valueY').between(F.col('b.startY'), F.col('b.endY')), 
    'left_anti'
).withColumn('isIn', F.lit(False))

result = dfa_joined.unionAll(dfb_joined).show()
+------+------+-----+
|valueX|valueY| isIn|
+------+------+-----+
|     1|     2| true|
|    -1|     4| true|
|     4|     3|false|
+------+------+-----+