如何在Databricks中实现两个Delta表的事务性写入？

原来的代码之所以满足不了原子性要求，是因为两次write操作是独立执行的——中间存在时间窗口，如果刚好有外部查询在这个窗口里执行，就会读到一个表更新、另一个没更新的错乱状态。要解决这个问题，得把两个表的更新塞进同一个事务里，确保要么两个表都更新成功，要么都保持原样，而且中间状态完全对外部不可见。

方案一：用Spark SQL事务块（最推荐） #

Delta Lake原生支持ACID事务，你可以用Spark SQL的BEGIN TRANSACTION和COMMIT把两个表的更新包起来，实现原子性。修改后的代码如下：

from pyspark.sql import SparkSession

# 初始化SparkSession
spark = SparkSession.builder.getOrCreate()

# 生成测试数据
table1_data = [(1, "A", 10.5, True),
               (2, "B", 20.5, False),
               (3, "C", 30.5, True),
               (4, "D", 40.5, False)]
table2_data = [(1, "E", 50.5, True),
               (2, "F", 60.5, False),
               (3, "G", 70.5, True),
               (4, "H", 80.5, False)]

# 创建DataFrame并注册临时视图
table1_df = spark.createDataFrame(table1_data, ["col1", "col2", "col3", "col4"])
table2_df = spark.createDataFrame(table2_data, ["col1", "col2", "col3", "col4"])
table1_df.createOrReplaceTempView("temp_table1")
table2_df.createOrReplaceTempView("temp_table2")

# 执行原子事务
spark.sql("""
BEGIN TRANSACTION;
-- 原子替换table1
CREATE OR REPLACE TABLE table1 AS SELECT * FROM temp_table1;
-- 原子替换table2
CREATE OR REPLACE TABLE table2 AS SELECT * FROM temp_table2;
COMMIT;
""")

方案说明： #

• BEGIN TRANSACTION启动一个事务，后续所有SQL操作都属于这个事务的一部分。
• CREATE OR REPLACE TABLE是原子性替换操作，事务提交前，外部查询只能看到表的旧版本。
• COMMIT会一次性提交两个表的更新，此时外部才能看到最新状态；如果中间任何一步失败，事务会自动回滚，两个表都保持原始状态。

方案二：用Python API结合事务上下文 #

如果你更习惯用Python API而非SQL，也可以用Delta Lake的事务API把两个写操作包裹在同一个事务里：

from pyspark.sql import SparkSession

# 初始化SparkSession（Databricks默认已配置Delta扩展，无需额外添加config）
spark = SparkSession.builder.getOrCreate()

# 生成测试数据
table1_data = [(1, "A", 10.5, True),
               (2, "B", 20.5, False),
               (3, "C", 30.5, True),
               (4, "D", 40.5, False)]
table2_data = [(1, "E", 50.5, True),
               (2, "F", 60.5, False),
               (3, "G", 70.5, True),
               (4, "H", 80.5, False)]

table1_df = spark.createDataFrame(table1_data, ["col1", "col2", "col3", "col4"])
table2_df = spark.createDataFrame(table2_data, ["col1", "col2", "col3", "col4"])

# 用事务包裹两个写操作
with spark.sql("BEGIN TRANSACTION").collect():
    # 写入table1
    table1_df.write.format("delta").mode("overwrite").option("overwriteSchema", "true").saveAsTable("table1")
    # 写入table2
    table2_df.write.format("delta").mode("overwrite").option("overwriteSchema", "true").saveAsTable("table2")
    # 提交事务
    spark.sql("COMMIT").collect()

方案说明： #

• with语句会自动管理事务生命周期：启动事务后执行两个写操作，最后自动提交。
• 只要其中一个写操作失败，整个事务就会回滚，两个表都不会被修改。
• 事务提交前，外部完全看不到中间的更新状态，彻底避免了数据错乱。

关键注意事项 #

• Databricks默认已集成Delta Lake，无需额外配置扩展（方案二中的config可省略）。
• 不要在事务中执行耗时过长的操作，事务持有的锁会影响其他查询的性能。
• 如果是增量更新（而非全量覆盖），可以用DeltaTable.merge()方法替代overwrite，只需将merge操作也放入事务即可。

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