能否在PySpark DataFrame上使用MERGE INTO？Databricks报错求助

你遇到的「找不到表」错误，核心原因是**MERGE INTO是Delta Lake专属的SQL语法，仅能作用于Delta表（或已注册到Spark元数据的临时/永久视图）**，而你当前的target和source只是内存中的DataFrame，SQL引擎无法直接识别它们。

以下是两种可行的合并方案：

方案一：转为Delta表后使用MERGE INTO（适合持久化+事务场景） #

如果需要持久化数据、支持增量更新或ACID事务，必须先将DataFrame转为Delta表，再执行MERGE INTO：

步骤1：将DataFrame转为Delta表并注册视图 #

# 将target保存为Delta表（持久化到DBFS路径）
target.write.format("delta").mode("overwrite").save("/tmp/target_delta")
# 注册为临时视图，供SQL调用
spark.sql("CREATE OR REPLACE TEMP VIEW target USING delta LOCATION '/tmp/target_delta'")

# 将source注册为临时视图（无需持久化，仅SQL引用）
source.createOrReplaceTempView("source")

步骤2：执行MERGE INTO语句 #

%sql
MERGE INTO target as t
USING source as s
ON t.id = s.id
WHEN MATCHED THEN 
  UPDATE SET *
WHEN NOT MATCHED THEN 
  INSERT *

查看合并结果 #

%sql
SELECT * FROM target ORDER BY id

方案二：用PySpark API实现内存合并（无需持久化） #

如果只是临时在内存中合并数据，不需要持久化，可以直接用PySpark API实现类似MERGE INTO的逻辑：

方式1：精准模拟UPSERT逻辑 #

# 1. 更新匹配的行：用source的数据覆盖target中同id的行
matched_updated = source.select("id", "name", "location", "contact")

# 2. 保留target中未匹配的行（source中没有的id）
unmatched_target = target.join(source, on="id", how="left_anti")

# 3. 添加source中新增的行（target中没有的id）
unmatched_source = source.join(target, on="id", how="left_anti")

# 4. 合并所有结果
final_df = matched_updated.union(unmatched_target).union(unmatched_source)

# 查看结果
final_df.orderBy("id").show()

方式2：简洁版去重合并 #

如果字段结构完全一致，可通过合并后去重实现（保留source的最新数据）：

# 合并两个DataFrame，按id去重时保留后出现的行（即source的数据）
combined_df = target.union(source)
final_df = combined_df.orderBy("id", ascending=False).dropDuplicates(["id"])

final_df.orderBy("id").show()

总结 #

• 若需要持久化数据、支持增量更新或事务，必须转为Delta表后使用MERGE INTO；
• 若仅需内存临时合并，直接用PySpark API即可，无需转为Delta表。

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