教程：跨表协调事务处理

重要

写入 Unity Catalog 托管 Delta 表的事务目前处于公共预览阶段。

写入 Unity Catalog 管理的 Iceberg 表的事务处于私密预览阶段。要加入此预览版，请提交管理式 Iceberg 表预览报名表。

在本教程中，你将使用这两种事务模式在 Azure Databricks 上跨多个语句和表协调更新：非交互式（BEGIN ATOMIC），会自动提交；以及交互式（BEGIN TRANSACTION），让你能够进行显式控制。本教程还演示如何将事务与存储过程和 SQL 脚本配合使用。

要求

Environment：访问Azure Databricks工作区。
计算：支持的计算类型因事务模式而异：
- 经典或无服务器 SQL 仓库支持这两种事务模式。
- 无服务器计算仅支持非交互式事务。
- 运行 Databricks Runtime 18.0 或更高版本的经典群集仅支持非交互式事务。
特权: CREATE TABLE 在 Unity Catalog 模式中。

设置示例表

写入多语句、多表事务的所有表必须：

是 Unity 目录托管表（Delta 或 Iceberg）
已启用目录提交

在 SQL 编辑器或笔记本中创建两个示例表：

-- Account data
CREATE TABLE IF NOT EXISTS sample_accounts (
  id INT,
  account_name STRING,
  balance DECIMAL(10,2)
) USING DELTA
TBLPROPERTIES (
  'delta.feature.catalogManaged' = 'supported'
);

-- Transaction records
CREATE TABLE IF NOT EXISTS sample_transactions (
  id INT,
  account_id INT,
  transaction_type STRING,
  amount DECIMAL(10,2)
) USING DELTA
TBLPROPERTIES (
  'delta.feature.catalogManaged' = 'supported'
);

注释

若要在现有表上启用事务，请运行：

ALTER TABLE <table_name> SET TBLPROPERTIES ('delta.feature.catalogManaged' = 'supported');

将示例数据插入两个表中：

INSERT INTO sample_accounts VALUES
  (1, 'Alice', 1000.00),
  (2, 'Bob', 500.00);

INSERT INTO sample_transactions VALUES
  (1, 1, 'deposit', 100.00);

验证设置：

SELECT * FROM sample_accounts;
SELECT * FROM sample_transactions;

输出：

sample_accounts:
id	account_name	balance
1	  Alice	        1000.00
2	  Bob	          500.00

sample_transactions:
id	account_id	transaction_type	amount
1	           1	         deposit	100.00

非交互式事务

非交互式事务使用 BEGIN ATOMIC ... END; 语法。所有语句作为单个原子单元运行。如果每个语句都成功，Azure Databricks自动提交。如果任何语句失败，Azure Databricks自动回滚所有更改。有关详细的语法和使用模式，请参阅非交互式事务。

执行成功的交易

以原子方式更新这两个表：

BEGIN ATOMIC
  -- Update Alice's account balance
  UPDATE sample_accounts
  SET balance = balance + 100.00
  WHERE id = 1;

  -- Record the deposit transaction
  INSERT INTO sample_transactions
  VALUES (2, 1, 'deposit', 100.00);
END;

验证 Alice 的余额现在为 1100.00：

SELECT * FROM sample_accounts WHERE id = 1;

验证两条事务记录现在是否存在：

SELECT * FROM sample_transactions;

余额更新和事务记录都是在一起创建的。如果任一语句失败，则两个更改都不会提交，Databricks 将终止事务而不产生副作用。

使用 SIGNAL 在满足某个条件时中止事务

当不满足用户定义的条件时，可以在SIGNAL区块中使用BEGIN ATOMIC ... END;来使事务失败。此示例插入一个余额为负的账户，然后在余额检查失败时使用 SIGNAL 使交易失败：

BEGIN ATOMIC
  INSERT INTO sample_accounts VALUES (3, 'Charlie', -50.00);

  IF (SELECT balance FROM sample_accounts WHERE id = 3) < 0 THEN
    SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'Account balance cannot be negative';
  END IF;
END;

SIGNAL 引发了一个错误，导致整个事务自动回滚。这会返回 0 行，因为插入操作已被回滚：

SELECT * FROM sample_accounts WHERE id = 3;

查看失败时自动回滚

运行一个事务，其中第一条语句有效，但第二条语句引用了一个不存在的表：

BEGIN ATOMIC
  -- Valid
  INSERT INTO sample_accounts VALUES (4, 'David', 300.00);
  -- Invalid
  INSERT INTO non_existent_table VALUES (1, 2, 3);
END;

交易失败，并显示错误信息。这会返回 0 行，因为整个事务都已回滚：

SELECT * FROM sample_accounts WHERE id = 4;

即使第一个 INSERT 语句有效，它也会回滚，因为第二个语句失败。这个示例说明了事务的全有或全无保证。

交互式事务

交互式事务让你具有对何时提交或回滚的显式控制。使用 BEGIN TRANSACTION 启动，然后 COMMIT 保存更改或 ROLLBACK 以放弃更改。

提交更改

启动事务：

BEGIN TRANSACTION;

进行更改（尚未提交）：

INSERT INTO sample_accounts VALUES (5, 'Eve', 850.00);
UPDATE sample_accounts SET balance = balance + 50.00 WHERE id = 2;

提交以使更改永久生效：

COMMIT;

确认 Eve 的账户现在已可见：

SELECT * FROM sample_accounts WHERE id = 5;

验证 Bob 的余额现为 550.00：

SELECT * FROM sample_accounts WHERE id = 2;

撤消更改

开始新事务：

BEGIN TRANSACTION;

进行更改：

INSERT INTO sample_accounts VALUES (6, 'Frank', 600.00);

验证该更改在当前会话中是否可见（在提交之前，该行对其他会话不可见）：

SELECT * FROM sample_accounts WHERE id = 6;

回滚以取消更改：

ROLLBACK;

这将返回零行，因为插入操作已回滚：

SELECT * FROM sample_accounts WHERE id = 6;

与存储过程和 SQL 脚本配合使用

可以将事务与存储过程合并，以创建可重用的事务逻辑。此模式适用于经常运行的复杂操作。

创建启用了目录提交的表

CREATE SCHEMA IF NOT EXISTS main.retail;

CREATE TABLE IF NOT EXISTS main.retail.orders (
  order_id STRING,
  customer_id STRING,
  amount DECIMAL(18,2)
) TBLPROPERTIES ('delta.feature.catalogManaged' = 'supported');

CREATE TABLE IF NOT EXISTS main.retail.orders_staging (
  order_id STRING,
  customer_id STRING,
  amount DECIMAL(18,2),
  batch_id STRING
) TBLPROPERTIES ('delta.feature.catalogManaged' = 'supported');

CREATE TABLE IF NOT EXISTS main.retail.total_sales (
  customer_id STRING,
  total_amount DECIMAL(18,2)
) TBLPROPERTIES ('delta.feature.catalogManaged' = 'supported');

定义存储过程

CREATE OR REPLACE PROCEDURE main.retail.apply_order(
    IN  p_order_id      STRING,
    IN  p_customer_id   STRING,
    IN  p_order_amount  DECIMAL(18,2)
)
LANGUAGE SQL
SQL SECURITY INVOKER
MODIFIES SQL DATA
AS
BEGIN
    -- Insert the order
    INSERT INTO main.retail.orders (order_id, customer_id, amount)
    VALUES (p_order_id, p_customer_id, p_order_amount);

    -- Update total sales per customer
    MERGE INTO main.retail.total_sales AS t
    USING (
        SELECT
          p_customer_id  AS customer_id,
          p_order_amount AS order_amount
    ) s
      ON t.customer_id = s.customer_id
    WHEN MATCHED THEN
      UPDATE SET t.total_amount = t.total_amount + s.order_amount
    WHEN NOT MATCHED THEN
      INSERT (customer_id, total_amount)
      VALUES (s.customer_id, s.order_amount);
END;

定义事务

BEGIN ATOMIC
    -- Staging batch id for this transaction
    DECLARE new_order_id STRING DEFAULT uuid();
    DECLARE v_batch_id STRING DEFAULT uuid();

    -- 1) Stage incoming customer and order rows
    INSERT INTO main.retail.orders_staging (order_id, customer_id, amount, batch_id)
    VALUES (new_order_id, 'CUST_123', 249.99, v_batch_id);

    -- 2) Drive final writes from staging to production via stored procedure
    FOR o AS
      SELECT
        order_id,
        customer_id,
        amount
      FROM main.retail.orders_staging
      WHERE batch_id = v_batch_id
    DO
        CALL main.retail.apply_order(
          o.order_id,
          o.customer_id,
          o.amount
        );
    END FOR;

    -- 3) Clean up processed staging rows
    DELETE FROM main.retail.orders_staging
    WHERE batch_id = v_batch_id;
END; -- 4) Commit the transaction

如果事务的任何部分失败，Databricks 将自动回滚所有更改。

清理

删除示例表：

DROP TABLE IF EXISTS sample_accounts;
DROP TABLE IF EXISTS sample_transactions;

DROP TABLE IF EXISTS main.retail.orders;
DROP TABLE IF EXISTS main.retail.orders_staging;
DROP TABLE IF EXISTS main.retail.total_sales;

其他资源

事务：事务支持概述。
事务模式：这两种模式的详细语法和模式。
目录提交：在您的表中启用事务支持。
使用来自不同客户端的事务：从 JDBC、ODBC 和Python应用程序运行事务。
ATOMIC 复合语句（非交互式事务）
BEGIN TRANSACTION （交互式事务）
COMMIT
ROLLBACK

反馈

此页面是否有帮助？

Last updated on 2026-06-24