PySpark更新Oracle数据库表遇阻，求高效可行解决方案

问题描述 #

我正在开展一个项目，需要使用PySpark更新Oracle数据库中的现有表，但当前采用的方法遇到了问题：

1. 尝试用spark.sql()执行更新查询时失败，因为该方法主要用于Spark内部的临时表或视图，无法直接操作外部Oracle表；且不能使用overwrite模式写入，否则会删除现有表，不符合要求。
2. 使用write.jdbc的append模式可正常插入数据，但无法执行更新操作。

相关代码片段：

if ids_not_in_target.count() > 0:
    ids_not_in_target.write.format("jdbc").mode("append").option("dbtable", target_table).option("url", target_jdbc_url).options(**target_oracle_properties).save()
else:
    update_sql = f"""
    UPDATE target_table
    SET column1 = value1,
        column2 = value2
    WHERE condition
    """
    spark.sql(update_sql)  # 这里无法正常工作

执行spark.sql()时出现错误：

[TABLE_OR_VIEW_NOT_FOUND] The table or view `<schema_name>`.`<table_name>` cannot be found. Verify the spelling and correctness of the schema and catalog.
 If you did not qualify the name with a schema, verify the current_schema() output, or qualify the name with the correct schema and catalog.
 To tolerate the error on drop use DROP VIEW IF EXISTS or DROP TABLE IF EXISTS.; line 2 pos 19;
 +- 'UnresolvedRelation [<schema>, <table_name>], [], false

注：数据库中该表确实存在，且可以正常读取和追加数据。

待解决问题 #

1. 使用PySpark更新Oracle表的最佳方案是什么？
2. 如何实现无需overwrite的高效更新方法？
3. 若无法直接用spark.sql()或write.jdbc更新，能否读取表数据到DataFrame中更新后仅写入变更内容？

解决方案 #

1. PySpark更新Oracle表的最佳方案 #

PySpark原生API不支持直接对JDBC外部表执行UPDATE操作，最佳方案是结合append模式处理新增数据，用Oracle原生SQL的MERGE INTO或直接UPDATE处理更新数据。其中MERGE INTO是最推荐的，因为它能批量处理插入和更新逻辑，减少数据库交互次数，效率更高。

2. 无需overwrite的高效更新方法 #

方案一：直接执行Oracle原生UPDATE语句

绕过spark.sql()，通过JDBC连接直接执行Oracle的更新语句，可使用Python的cx_Oracle库或Spark的JDBC接口实现：

# 方式1：用cx_Oracle直接连接执行
import cx_Oracle

# 从配置中提取连接信息
user = target_oracle_properties["user"]
password = target_oracle_properties["password"]
dsn = target_jdbc_url.split("jdbc:oracle:thin:@")[1]

conn = cx_Oracle.connect(user=user, password=password, dsn=dsn)
cursor = conn.cursor()

# 单条更新
update_sql = """
UPDATE target_table
SET column1 = :val1, column2 = :val2
WHERE id = :target_id
"""
cursor.execute(update_sql, val1=value1, val2=value2, target_id=target_id)

# 批量更新（如果有多个更新项）
update_data = [(val1_1, val2_1, id1), (val1_2, val2_2, id2)]
cursor.executemany(update_sql, update_data)

conn.commit()
cursor.close()
conn.close()

# 方式2：用Spark JDBC接口执行原生SQL
spark.read.jdbc(
    url=target_jdbc_url,
    table=f"({update_sql}) as tmp_update",
    properties=target_oracle_properties
)

方案二：用MERGE INTO批量处理插入+更新

如果需要同时处理新增和更新，MERGE INTO是最高效的方式，步骤如下：

1. 将待处理的DataFrame写入Oracle临时表（推荐用全局临时表，会话结束后自动清理）。
2. 执行MERGE INTO语句，通过主键匹配目标表和临时表，自动判断是插入新数据还是更新已有数据。

示例代码：

# 1. 将待处理数据写入Oracle全局临时表
temp_table = "GLOBAL_TEMP.TMP_UPDATE_DATA"
your_data_df.write.format("jdbc")\
    .mode("overwrite")\
    .option("dbtable", temp_table)\
    .option("url", target_jdbc_url)\
    .options(**target_oracle_properties)\
    .save()

# 2. 执行MERGE INTO语句
merge_sql = f"""
MERGE INTO {target_table} t
USING {temp_table} s
ON (t.id = s.id)  -- 替换为你的主键匹配条件
WHEN MATCHED THEN
  UPDATE SET 
    t.column1 = s.column1,
    t.column2 = s.column2
WHEN NOT MATCHED THEN
  INSERT (id, column1, column2)
  VALUES (s.id, s.column1, s.column2)
"""

# 通过Spark JDBC执行MERGE语句
spark.read.jdbc(
    url=target_jdbc_url,
    table=f"({merge_sql}) as tmp_merge",
    properties=target_oracle_properties
)

3. 读取表到DataFrame更新后仅写入变更内容 #

可以实现，但需注意数据量，避免全量读取影响性能，步骤如下：

1. 读取目标表的主键和需要对比的字段到DataFrame（无需全量读取所有字段）。
2. 与待更新的数据进行JOIN，筛选出主键匹配且字段有变化的记录。
3. 将变更数据写入临时表，再通过原生SQL更新目标表，或用MERGE INTO处理。

示例代码：

# 1. 读取目标表的必要字段（主键+需更新字段）
target_df = spark.read.jdbc(
    url=target_jdbc_url,
    table=f"(SELECT id, column1, column2 FROM {target_table}) as target_subset",
    properties=target_oracle_properties
)

# 2. 筛选出需要更新的记录（假设id是主键）
updated_records = your_updated_df.join(target_df, on="id", how="inner")\
    .filter(
        (your_updated_df.column1 != target_df.column1) | 
        (your_updated_df.column2 != target_df.column2)
    )\
    .select(your_updated_df["id"], your_updated_df["column1"], your_updated_df["column2"])

# 3. 将变更数据写入临时表
temp_table = "GLOBAL_TEMP.TMP_CHANGES"
updated_records.write.format("jdbc")\
    .mode("overwrite")\
    .option("dbtable", temp_table)\
    .option("url", target_jdbc_url)\
    .options(**target_oracle_properties)\
    .save()

# 4. 执行更新SQL
update_sql = f"""
UPDATE {target_table} t
SET 
  column1 = (SELECT column1 FROM {temp_table} s WHERE s.id = t.id),
  column2 = (SELECT column2 FROM {temp_table} s WHERE s.id = t.id)
WHERE EXISTS (SELECT 1 FROM {temp_table} s WHERE s.id = t.id)
"""

spark.read.jdbc(
    url=target_jdbc_url,
    table=f"({update_sql}) as tmp_update",
    properties=target_oracle_properties
)

如果目标表数据量极大，建议通过时间戳或其他条件读取增量数据，而非全量读取，提升效率。

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