PySpark：基于另一DataFrame修改列值的实现咨询

核心思路 #

通过**左连接（Left Join）**关联两个DataFrame，结合条件判断函数when实现前缀逻辑，这是Spark中最直观且性能最优的方案——Spark的优化器会对Join操作进行针对性优化，同时条件判断属于窄依赖操作，不会引入额外的Shuffle开销。

代码实现（Python） #

首先初始化示例数据：

from pyspark.sql import SparkSession
from pyspark.sql.functions import when, concat, lit, col, broadcast

spark = SparkSession.builder.appName("LicensePrefixProcessing").getOrCreate()

# 第一个DataFrame（id、license_no）
df1 = spark.createDataFrame([
    (2005, "1011"),
    (2006, "1022"),
    (2007, "3911"),
    (2008, "4000")  # 新增测试数据：license_no不存在于df2
], ["id", "license_no"])

# 第二个DataFrame（license_no、active）
df2 = spark.createDataFrame([
    ("1011", "y"),
    ("1022", "y"),
    ("3911", "n")
], ["license_no", "active"])

然后执行核心逻辑：

result_df = df1.join(broadcast(df2), on="license_no", how="left") \
    .withColumn("new_id", 
                when(col("active") == "y", concat(lit("99"), col("id").cast("string")))
                .otherwise(concat(lit("88"), col("id").cast("string")))) \
    .withColumn("new_license_no", 
                when(col("active") == "y", concat(lit("99"), col("license_no")))
                .otherwise(concat(lit("88"), col("license_no")))) \
    .select("new_id", "new_license_no")

# 查看结果
result_df.show()

执行结果：

+------+---------------+
|new_id|new_license_no |
+------+---------------+
|992005|991011         |
|992006|991022         |
|882007|883911         |
|882008|884000         |
+------+---------------+

关键细节说明 #

1. Broadcast优化：如果df2数据量较小，使用broadcast()将其广播到所有Executor节点，避免Shuffle操作，大幅提升Join性能。
2. 类型转换：由于id是数值类型，需用cast("string")转换为字符串后再与前缀拼接，否则会触发数值运算（99+2005=2104，不符合需求）。
3. 条件覆盖：otherwise分支会自动覆盖两种情况：license_no不存在于df2，或存在但active不为'y'。

能否单条命令实现？ #

可以。上述逻辑可以合并为一条链式调用（可读性稍差，但属于单条命令范畴）：

result_df = df1.join(broadcast(df2), on="license_no", how="left").withColumn("new_id", when(col("active") == "y", concat(lit("99"), col("id").cast("string"))).otherwise(concat(lit("88"), col("id").cast("string")))).withColumn("new_license_no", when(col("active") == "y", concat(lit("99"), col("license_no"))).otherwise(concat(lit("88"), col("license_no")))).select("new_id", "new_license_no")

补充：Scala版本实现 #

如果使用Scala，逻辑完全一致：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{when, concat, lit, col, broadcast}

val spark = SparkSession.builder.appName("LicensePrefixProcessing").getOrCreate()

val df1 = spark.createDataFrame(Seq(
    (2005, "1011"),
    (2006, "1022"),
    (2007, "3911"),
    (2008, "4000")
)).toDF("id", "license_no")

val df2 = spark.createDataFrame(Seq(
    ("1011", "y"),
    ("1022", "y"),
    ("3911", "n")
)).toDF("license_no", "active")

val resultDf = df1.join(broadcast(df2), Seq("license_no"), "left")
    .withColumn("new_id", when(col("active") === "y", concat(lit("99"), col("id").cast("string"))).otherwise(concat(lit("88"), col("id").cast("string"))))
    .withColumn("new_license_no", when(col("active") === "y", concat(lit("99"), col("license_no"))).otherwise(concat(lit("88"), col("license_no"))))
    .select("new_id", "new_license_no")

resultDf.show()

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