如何将含内连接的SQL UPDATE查询转换为PySpark实现？

原Access SQL查询 #

你给出的两条Access更新查询如下：

第一条：

UPDATE EMPLOYEE INNER JOIN [DEPARTMENT] ON
EMPLOYEE.STATEPROVINCE = [DEPARTMENT].[STATE_LEVEL] 
SET EMPLOYEE.STATEPROVINCE = [DEPARTMENT]![STATE_ABBREVIATION];

第二条：

UPDATE EMPLOYEE INNER JOIN [DEPARTMENT] ON
EMPLOYEE.STATEPROVINCE = [DEPARTMENT].[STATE_LEVEL] 
SET EMPLOYEE.MARKET = [DEPARTMENT]![MARKET];

由于PySpark没有直接支持UPDATE ... JOIN的语法，我们可以通过两种方式实现相同逻辑：

方法一：DataFrame 连接+覆盖写入 #

这种方式适合批量更新，先通过内连接获取需要更新的字段，生成新的DataFrame后覆盖原表。

步骤1：读取源表 #

首先读取Employee和Department表（根据实际存储格式调整读取方式）：

# 示例：从Hive表读取，也可以用read.csv/read.parquet等
employee_df = spark.read.table("EMPLOYEE")
department_df = spark.read.table("DEPARTMENT")

步骤2：执行更新逻辑 #

可以分开处理两个更新，也可以合并成一次操作（更高效）：

合并更新两个字段

# 内连接后替换目标字段，保留其他原有字段
updated_employee_df = employee_df.join(
    department_df,
    employee_df.STATEPROVINCE == department_df.STATE_LEVEL,
    "inner"
).select(
    # 替换STATEPROVINCE为部门的缩写
    department_df.STATE_ABBREVIATION.alias("STATEPROVINCE"),
    # 替换MARKET为部门的MARKET
    department_df.MARKET.alias("MARKET"),
    # 保留Employee表的其他字段，比如主键、姓名等（根据实际表结构补充）
    employee_df.EMPLOYEE_ID,
    employee_df.NAME,
    # ...其他字段
)

# 覆盖写入原表，mode="overwrite"会替换整个表
updated_employee_df.write.mode("overwrite").saveAsTable("EMPLOYEE")

分开处理单个字段更新

如果需要单独更新某一个字段，比如先更新STATEPROVINCE：

updated_state_df = employee_df.join(
    department_df,
    employee_df.STATEPROVINCE == department_df.STATE_LEVEL,
    "inner"
).select(
    department_df.STATE_ABBREVIATION.alias("STATEPROVINCE"),
    employee_df.EMPLOYEE_ID,
    employee_df.NAME,
    employee_df.MARKET,
    # ...其他字段
)

updated_state_df.write.mode("overwrite").saveAsTable("EMPLOYEE")

再更新MARKET：

updated_market_df = employee_df.join(
    department_df,
    employee_df.STATEPROVINCE == department_df.STATE_LEVEL,
    "inner"
).select(
    employee_df.STATEPROVINCE,
    employee_df.EMPLOYEE_ID,
    employee_df.NAME,
    department_df.MARKET.alias("MARKET"),
    # ...其他字段
)

updated_market_df.write.mode("overwrite").saveAsTable("EMPLOYEE")

方法二：Spark SQL MERGE INTO（Spark 2.4+支持） #

如果需要增量更新（仅修改匹配的行，不覆盖整个表），可以使用MERGE INTO语法，这更接近Access的UPDATE逻辑。

步骤1：注册临时视图 #

employee_df.createOrReplaceTempView("EMPLOYEE")
department_df.createOrReplaceTempView("DEPARTMENT")

步骤2：执行MERGE语句 #

合并更新两个字段

MERGE INTO EMPLOYEE e
USING DEPARTMENT d
ON e.STATEPROVINCE = d.STATE_LEVEL
WHEN MATCHED THEN
  UPDATE SET 
    e.STATEPROVINCE = d.STATE_ABBREVIATION,
    e.MARKET = d.MARKET;

单独更新单个字段

更新STATEPROVINCE：

MERGE INTO EMPLOYEE e
USING DEPARTMENT d
ON e.STATEPROVINCE = d.STATE_LEVEL
WHEN MATCHED THEN
  UPDATE SET e.STATEPROVINCE = d.STATE_ABBREVIATION;

更新MARKET：

MERGE INTO EMPLOYEE e
USING DEPARTMENT d
ON e.STATEPROVINCE = d.STATE_LEVEL
WHEN MATCHED THEN
  UPDATE SET e.MARKET = d.MARKET;

注意事项 #

• 确保Employee表有唯一标识字段（如EMPLOYEE_ID），避免连接时出现重复匹配导致数据异常
• 使用overwrite模式时会替换整个表，适合全量更新；MERGE INTO仅更新匹配行，适合增量场景
• 读取和写入表的方式需根据实际存储系统调整（如HDFS、S3、本地文件等）

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