PySpark中如何通过join操作添加或更新DataFrame列值

需求是基于另一个DataFrame（记为ds）的等值匹配条件，更新目标DataFrame（记为df）的列值，以下代码可直接复制到Notebook中复现问题：

构造目标DataFrame df的代码：

data = [["101", "sravan", "vignan"],
        ["102", "ramya", "vvit"],
        ["103", "rohith", "klu"],
        ["104", "sridevi", "vignan"],
        ["105", "gnanesh", "iit"]]
columns = ['rollNo', 'name', 'lastName']
df = spark.createDataFrame(data=data, schema=columns, verifySchema=True)

构造匹配用DataFrame ds的代码：

ds = spark.createDataFrame(
    [
        ("100", "2015-01-01", "2015-01-01T13:51:39.340396Z"),
        ("101", "2015-01-01", "2015-01-01T12:14:58.597216Z"),
        ("102", "2015-01-01", "2015-01-01T13:51:40.417052Z"),
        ("103", "2015-01-01", "2015-01-01T13:51:40.519832Z"),
        ("107", "2015-01-01", "2015-01-01T13:51:40.519832Z"),
        ("108", "2015-01-01", "2015-01-01T13:51:40.519832Z"),
        ("109", "2015-01-01", "2015-01-01T13:51:40.519832Z"),
    ],
    ["rollNo", "creation_date", "last_update_time"]
)

执行leftanti连接生成dt，提取左表df中与ds无匹配rollNo的行：

dt = df.join(ds, ["rollNo"], "leftanti")
dt.show(5,False)
df.show(5,False)

dt的运行结果如下，仅包含rollNo为104、105的两条记录：

+------+-------+--------+
|rollNo|name   |lastName|
+------+-------+--------+
|104   |sridevi|vignan  |
|105   |gnanesh|iit     |
+------+-------+--------+

原始df共5条记录：

+------+-------+--------+
|rollNo|name   |lastName|
+------+-------+--------+
|101   |sravan |vignan  |
|102   |ramya  |vvit    |
|103   |rohith |klu     |
|104   |sridevi|vignan  |
|105   |gnanesh|iit     |
+------+-------+--------+

尝试通过以下代码为df新增is_deleted列，逻辑为rollNo存在于dt中的行赋值为True，其余赋值为False，但运行后所有行的is_deleted均为True，不符合预期：

df.withColumn('is_deleted', when(dt.rollNo == df.rollNo, True).otherwise(False)).show(5,False) 

实际错误运行结果：

+------+-------+--------+----------+
|rollNo|name   |lastName|is_deleted|
+------+-------+--------+----------+
|101   |sravan |vignan  |true      |
|102   |ramya  |vvit    |true      |
|103   |rohith |klu     |true      |
|104   |sridevi|vignan  |true      |
|105   |gnanesh|iit     |true      |
+------+-------+--------+----------+

预期运行结果：rollNo为101/102/103的行is_deleted为False，rollNo为104/105的行is_deleted为True：

+------+-------+--------+----------+
|rollNo|name   |lastName|is_deleted|
+------+-------+--------+----------+
|101   |sravan |vignan  |false     |
|102   |ramya  |vvit    |false     |
|103   |rohith |klu     |false     |
|104   |sridevi|vignan  |true      |
|105   |gnanesh|iit     |true      |
+------+-------+--------+----------+

直接在withColumn中引用未做关联的另一个DataFrame的列做判断是错误写法。跨DataFrame直接引用列不会触发你预期的逐行等值匹配，反而会产生隐式笛卡尔积，加上空值判断的逻辑偏差，最终导致所有行的计算结果都为True。

• 方法1：基于已生成的dt做左连接匹配
  给dt表先打上删除标记，再和原df做左连接，未匹配到标记的行填充为False即可：

from pyspark.sql.functions import col, lit, when

# 为dt中的行增加删除标记
dt_marked = dt.select("rollNo", lit(True).alias("is_deleted"))
# 左连接后补全未匹配行的标记值
result = df.join(dt_marked, on="rollNo", how="left")\
    .withColumn("is_deleted", when(col("is_deleted").isNull(), lit(False)).otherwise(col("is_deleted")))

result.show(5, False)

• 方法2：跳过dt生成步骤，直接关联ds打标（性能更优）
  不需要提前生成dt，直接关联ds判断rollNo是否存在即可，少一次join计算性能更好：

from pyspark.sql.functions import col, lit, when

# 给ds的rollNo打存在标记
ds_marked = ds.select("rollNo", lit(True).alias("exist_flag"))
# 左连接后，不存在标记的行即为已删除
result = df.join(ds_marked, on="rollNo", how="left")\
    .withColumn("is_deleted", when(col("exist_flag").isNull(), lit(True)).otherwise(lit(False)))\
    .drop("exist_flag")

result.show(5, False)

两种方法运行后都会输出符合预期的结果。

注意：Spark中不同DataFrame之间的列匹配，必须先通过join操作完成关联，不能直接在withColumn、select等算子中跨DataFrame引用列做判断，否则会出现不可预期的计算错误。

内容的提问来源于stack exchange，提问作者user1119283