Руководство. Координация транзакций между таблицами

Это важно

Транзакции, записываемые в управляемые таблицы Iceberg каталога Unity, находятся в закрытой предварительной версии. Чтобы присоединиться к этой предварительной версии, отправьте форму регистрации на предварительный просмотр управляемых таблиц Iceberg.

В этом руководстве вы используете оба режима транзакций для координации обновлений в нескольких операторах и таблицах в Azure Databricks: неинтерактивный (BEGIN ATOMIC), который автоматически фиксирует изменения, и интерактивный (BEGIN TRANSACTION), который позволяет явно управлять транзакцией. В этом руководстве также показано, как использовать транзакции с хранимыми процедурами и скриптами SQL.

Требования

Environment: доступ к рабочей области Azure Databricks.
Вычисления: поддерживаемые типы вычислений зависят от режима транзакции:
- Классическое или бессерверное хранилище SQL поддерживает оба режима транзакций.
- Бессерверные вычисления поддерживают только неинтерактивные транзакции.
- Классические кластеры под управлением Databricks Runtime 18.0 или более поздней версии поддерживают только неинтерактивные транзакции.
Привилегии: CREATE TABLE в схеме каталога Unity .

Настройка примеров таблиц

Все таблицы, записанные в многооперационную, многотабличную транзакцию, должны:

Управляемые таблицы каталога 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;

Убедитесь, что баланс Алисы теперь равен 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 вызывает программную ошибку, в результате чего вся транзакция автоматически откатывается. Это возвращает ноль строк, так как операция вставки была отменена:

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 для запуска, а затем ФИКСАЦИЯ для сохранения изменений или ROLLBACK для их отмены.

Фиксация изменений

Запуск транзакции:

BEGIN TRANSACTION;

Внесите изменения (пока не утверждены):

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

Принять чтобы сделать изменения постоянными.

COMMIT;

Убедитесь, что учетная запись Ева теперь отображается:

SELECT * FROM sample_accounts WHERE id = 5;

Убедитесь, что баланс Боба сейчас 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