首先执行left-join，然后使用简单的when-otherwise构造，如下所示：

from pyspark.sql.functions import col, when

dfA = spark.createDataFrame([("John Smith", 93818, "Bad Math"), ("Jane Doe", 91982, "Invalid Row")], "Name: string, TransactionID: int, Alert_flag: string")
dfB = spark.createDataFrame([("Bad Math", "Incorrect Calculations"), ("Invalid Row", "Invalid Transaction")], "Alert_flag_OLD: string, Alert_flag_NEW: string")

dfC = dfA.join(dfB, col("Alert_flag") == col("Alert_flag_OLD"), "left")
dfC.withColumn("Alert_flag", when(col("Alert_flag_NEW").isNotNull(), col("Alert_flag_NEW"))).select(["Name", "TransactionID", "Alert_flag"]).show()

+----------+-------------+----------------------+
|Name      |TransactionID|Alert_flag            |
+----------+-------------+----------------------+
|John Smith|93818        |Incorrect Calculations|
|Jane Doe  |91982        |Invalid Transaction   |
+----------+-------------+----------------------+