Kurz: Koordinace transakcí napříč tabulkami

Důležité

Transakce, které se zapisují do spravovaných tabulek Delta katalogu Unity, jsou ve verzi Public Preview.

Transakce, které zapisují do katalogu Unity spravované tabulky Iceberg, jsou ve verzi Private Preview. Pokud se chcete připojit k této verzi Preview, odešlete formulář pro registraci spravovaných tabulek Iceberg ve verzi Preview.

V tomto kurzu použijete oba režimy transakcí ke koordinaci aktualizací mezi více příkazy a tabulkami v Azure Databricks: neinteraktivní (BEGIN ATOMIC), které se automaticky potvrdí a interaktivní (BEGIN TRANSACTION), což vám dává explicitní řízení. Tento kurz také ukazuje, jak používat transakce s uloženými procedurami a skriptováním SQL.

Požadavky

  • Environment: Přístup k pracovnímu prostoru Azure Databricks.
  • Výpočty: Podporované typy výpočetních prostředků se liší podle režimu transakcí:
  • Oprávnění: CREATE TABLE ve schématu Unity Catalog.

Nastavení ukázkových tabulek

Všechny tabulky, které jsou zapsány ve více příkazech v rámci transakce s více tabulkami, musí:

V editoru SQL nebo poznámkovém bloku vytvořte dvě ukázkové tabulky:

-- 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'
);

Note

Pokud chcete povolit transakce v existující tabulce, spusťte:

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

Vložte ukázková data do obou tabulek:

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

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

Ověřte nastavení:

SELECT * FROM sample_accounts;
SELECT * FROM sample_transactions;

Výstup:

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

Neinteraktivní transakce

Neinteraktivní transakce používají BEGIN ATOMIC ... END; syntaxi. Všechny příkazy běží jako jedna atomická jednotka. Pokud je každý příkaz úspěšný, Azure Databricks potvrdí automaticky. Pokud některý příkaz selže, Azure Databricks vrátí všechny změny automaticky. Podrobné vzorce syntaxe a použití najdete v neinteraktivních transakcích.

Spuštění úspěšné transakce

Aktualizujte atomickým způsobem obě tabulky:

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;

Ověřte, že zůstatek Alice je teď 1100,00:

SELECT * FROM sample_accounts WHERE id = 1;

Ověřte, že teď existují dva záznamy transakcí:

SELECT * FROM sample_transactions;

Aktualizace zůstatku i záznam transakce byly vytvořeny společně. Pokud by některý z příkazů selhal, nebyla by potvrzena žádná změna a Databricks by ukončil transakci bez vedlejších účinků.

Použijte SIGNAL k selhání transakce na základě podmínky

Uvnitř bloku SIGNAL můžete použít BEGIN ATOMIC ... END; k selhání transakce, pokud není splněna podmínka definovaná uživatelem. Tento příklad vloží účet se záporným zůstatkem a pak použije SIGNAL k selhání transakce, pokud kontrola zůstatku selže:

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 vyvolá chybu, která způsobí, že je celá transakce automaticky zrušena. To vrátí nula řádků, protože operace INSERT byla vrácena zpět:

SELECT * FROM sample_accounts WHERE id = 3;

Zobrazení automatického vrácení zpět při selhání

Spusťte transakci, kde je první příkaz platný, ale druhý odkazuje na tabulku, která neexistuje:

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

Transakce selže s chybou. Vrátí se 0 řádků, protože celá transakce byla vrácena zpět:

SELECT * FROM sample_accounts WHERE id = 4;

I když byl první INSERT příkaz platný, vrátil se zpět, protože druhý příkaz selhal. To ukazuje záruku transakcí typu vše nebo nic.

Interaktivní transakce

Interaktivní transakce poskytují explicitní kontrolu nad tím, kdy se mají potvrdit nebo vrátit zpět. Pomocí BEGIN TRANSACTION spuštěte transakci, poté použijte COMMIT k uložení změn nebo ROLLBACK k jejich zrušení.

Potvrzení změn

Zahájení transakce:

BEGIN TRANSACTION;

Proveďte změny (dosud nepotvrzené):

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

Potvrďte, že provedete trvalé změny:

COMMIT;

Ověřte, že je účet Eve viditelný:

SELECT * FROM sample_accounts WHERE id = 5;

Ověřte, že Bobův zůstatek je nyní 550.00:

SELECT * FROM sample_accounts WHERE id = 2;

Vrátit změny

Spusťte novou transakci:

BEGIN TRANSACTION;

Proveďte změnu:

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

Ověřte, zda je změna viditelná v rámci vaší relace (řádek není viditelný v jiných relacích, dokud nejsou změny potvrzeny):

SELECT * FROM sample_accounts WHERE id = 6;

Vraťte se zpět a zahoďte změnu:

ROLLBACK;

To vrátí nula řádků, protože operace INSERT byla vrácena zpět:

SELECT * FROM sample_accounts WHERE id = 6;

Použití s uloženými procedurami a skriptováním SQL

Transakce můžete kombinovat s uloženými procedurami a vytvořit opakovaně použitelnou logiku transakce. Tento model je užitečný pro složité operace, které spouštíte často.

  1. Vytvoření tabulek s povolenými potvrzeními katalogu

    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');

  2. Definování uložené procedury

    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;

  3. Definování transakce

    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

Pokud jakákoli část transakce selže, Databricks vrátí zpět všechny změny automaticky.

Uklidit

Odeberte ukázkové tabulky:

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;

Další zdroje informací

  • Last updated on