PySpark执行MERGE INTO语句报错：暂不支持该操作

你遇到的MERGE INTO TABLE is not supported temporarily错误，核心原因是Spark的MERGE INTO操作仅支持具备ACID事务能力的物理表，无法直接用于临时视图。你当前创建的current和updates是视图，不符合MERGE INTO的要求。

以下是几种可行的解决办法：

1. 使用Delta Lake（推荐） #

Delta Lake是Spark官方适配的ACID存储层，原生支持MERGE INTO操作，无需依赖Hive：

• 安装Delta Lake依赖：

  pip install delta-spark
  

• 初始化带Delta配置的SparkSession：

  from pyspark.sql import SparkSession
  from delta.tables import DeltaTable
  
  spark = SparkSession.builder \
      .appName("MergeWithDelta") \
      .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
      .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
      .getOrCreate()
  

• 将原有视图转换为Delta物理表：

  # 把current视图写入Delta表
  current.write.format("delta").mode("overwrite").save("./current_delta")
  # 把updates视图写入Delta表
  updates.write.format("delta").mode("overwrite").save("./updates_delta")
  

• 执行MERGE INTO语句（直接引用Delta表路径）：

  MERGE INTO delta.`./current_delta` current
  USING (
      SELECT updates.Name as mergeKey, updates.*
      FROM delta.`./updates_delta` updates
      UNION ALL
      SELECT NULL as mergeKey, updates.*
      FROM delta.`./updates_delta` updates 
      JOIN delta.`./current_delta` current
      ON updates.Name = current.Name 
      WHERE current.current = true 
  ) staged_updates
  ON current.Name = mergeKey
  WHEN MATCHED AND current.current = true THEN  
      UPDATE SET current = false, validity_end = CURRENT_TIMESTAMP()    
  WHEN NOT MATCHED THEN 
      INSERT(S_No,Name, DOB, validity_start, validity_end, current) 
      VALUES(null,staged_updates.Name, staged_updates.updated_DOB, current_timestamp(), null, true)
  

2. 使用Hive事务表 #

如果你的环境已配置Hive，可以创建支持ACID的Hive表：

• 创建事务型Hive表：

  CREATE TABLE current (
      S_No INT,
      Name STRING,
      DOB DATE,
      validity_start TIMESTAMP,
      validity_end TIMESTAMP,
      current BOOLEAN
  )
  STORED AS ORC
  TBLPROPERTIES ('transactional'='true');
  

• 将原有视图数据插入到该物理表后，再执行你的MERGE INTO语句。

3. 手动实现合并逻辑（无额外依赖） #

如果不想引入第三方库，可以拆分MERGE逻辑为更新+插入两个步骤：

1. 生成需要更新的记录：

   update_rows = spark.sql("""
       SELECT 
           current.S_No, 
           current.Name, 
           current.DOB, 
           current.validity_start, 
           CURRENT_TIMESTAMP() as validity_end, 
           false as current
       FROM current
       JOIN updates ON current.Name = updates.Name
       WHERE current.current = true
   """)
   

2. 生成需要插入的新记录：

   insert_rows = spark.sql("""
       SELECT 
           null as S_No, 
           updates.Name, 
           updates.updated_DOB as DOB, 
           CURRENT_TIMESTAMP() as validity_start, 
           null as validity_end, 
           true as current
       FROM updates
       LEFT JOIN current ON updates.Name = current.Name
       WHERE current.Name IS NULL
   """)
   

3. 合并所有需要保留的记录并覆盖原表：

   # 保留未被更新的有效记录
   remaining_rows = spark.sql("""
       SELECT * FROM current
       WHERE current = true AND Name NOT IN (SELECT Name FROM updates)
   """)
   # 合并所有数据并写入物理表
   final_data = update_rows.union(insert_rows).union(remaining_rows)
   final_data.write.mode("overwrite").saveAsTable("current_physical")
   

• MERGE INTO仅支持ACID物理表，临时视图、内存级DataFrame均不支持该操作。
• Delta Lake是最轻量的解决方案，无需依赖Hive集群，适合大多数开发场景。

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