Tryby transakcji

Transakcje na Azure Databricks obsługują dwa tryby:

• Nieinterakcyjne (BEGIN ATOMIC): automatycznie zatwierdza lub cofa. Najlepiej nadaje się do zaplanowanych zadań i ustalonych sekwencji instrukcji.
• Interakcyjne (BEGIN TRANSACTION): zapewnia ręczną kontrolę nad zatwierdzaniem i wycofywaniem. Najlepsze do walidacji, debugowania i klientów JDBC.

Aby zapoznać się z wymaganiami i omówieniem transakcji, zobacz Transakcje. Aby zapoznać się z praktycznymi ćwiczeniami w obu trybach, zobacz Samouczek: koordynowanie transakcji między tabelami.

Transakcje nieinterakcyjne

Transakcje nieinterakcyjne używają skryptów SQL ze ATOMIC słowem kluczowym . Blok instrukcji złożonej ATOMIC uruchamia wszystkie instrukcje jako pojedynczą jednostkę atomową. Wszyscy razem odnoszą sukces lub wszyscy razem ponoszą porażkę.

Obsługiwane zasoby obliczeniowe: dowolny magazyn SQL,bezserwerowe zasoby obliczeniowe lub klaster z uruchomionym środowiskiem Databricks Runtime 18.0 lub nowszym.

Obsługiwana składnia: obsługuje bloki SQL, Scala spark.sql i bloki PySpark spark.sql .

Składnia

BEGIN ATOMIC
  statement1;
  statement2;
  statement3;
END;

Azure Databricks automatycznie zatwierdza wszystkie zmiany, jeśli wszystkie polecenia zakończą się sukcesem. Jeśli jakakolwiek instrukcja nie powiedzie się, Azure Databricks automatycznie wycofa wszystkie zmiany.

Używanie w edytorze SQL

Uruchamiaj transakcje nieinterakcyjne bezpośrednio w edytorze SQL. Wybierz cały blok instrukcji złożonej ATOMIC i uruchom go jako pojedynczą instrukcję:

BEGIN ATOMIC
  DELETE FROM staging_sales WHERE load_date < current_date() - INTERVAL 7 DAYS;

  INSERT INTO staging_sales
  SELECT * FROM raw_sales WHERE load_date = current_date();

  MERGE INTO sales AS target
  USING staging_sales AS source
  ON target.sale_id = source.sale_id
  WHEN MATCHED THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *;
END;

Używanie w laptopach

Uruchamianie transakcji nieinterakcyjnych w notesach przy użyciu komórek SQL lub programowych interfejsów API.

SQL

BEGIN ATOMIC
  UPDATE inventory SET quantity = quantity - 10 WHERE product_id = 2001;
  UPDATE inventory SET quantity = quantity + 10 WHERE product_id = 2002;
  INSERT INTO inventory_moves (from_product, to_product, quantity, move_date)
  VALUES (2001, 2002, 10, current_date());
END;

Python

spark.sql("""
BEGIN ATOMIC
  UPDATE inventory SET quantity = quantity - 10 WHERE product_id = 2001;
  UPDATE inventory SET quantity = quantity + 10 WHERE product_id = 2002;
  INSERT INTO inventory_moves (from_product, to_product, quantity, move_date)
  VALUES (2001, 2002, 10, current_date());
END;
""")

Scala

spark.sql("""
BEGIN ATOMIC
  UPDATE inventory SET quantity = quantity - 10 WHERE product_id = 2001;
  UPDATE inventory SET quantity = quantity + 10 WHERE product_id = 2002;
  INSERT INTO inventory_moves (from_product, to_product, quantity, move_date)
  VALUES (2001, 2002, 10, current_date());
END;
""")

Używanie w zaplanowanych zadaniach

Transakcje nieinterakcyjne działają dobrze w zaplanowanych zadaniach, ponieważ automatycznie obsługują zatwierdzanie i wycofywanie:

BEGIN ATOMIC
  -- Clear previous staging data
  DELETE FROM staging_daily_sales WHERE load_date = current_date();

  -- Load new data
  INSERT INTO staging_daily_sales
  SELECT sale_id, customer_id, amount, sale_date, current_date() as load_date
  FROM raw_sales
  WHERE sale_date = current_date() - INTERVAL 1 DAY;

  -- Validate row count (fails transaction if no data)
  IF (SELECT COUNT(*) FROM staging_daily_sales WHERE load_date = current_date()) = 0 THEN
    SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'No sales data loaded for yesterday';
  END IF;

  -- Merge into production
  MERGE INTO daily_sales AS target
  USING staging_daily_sales AS source
  ON target.sale_id = source.sale_id
  WHEN MATCHED THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *;
END;

Jeśli jakakolwiek instrukcja nie powiedzie się, w tym asercji, cała transakcja zostanie wycofana automatycznie.

Używanie z funkcją JDBC

Klienci zewnętrzni mogą uruchamiać transakcje nieinterakcyjne.

JDBC

String sql = """
    BEGIN ATOMIC
      INSERT INTO orders (order_id, total) VALUES (1001, 500.00);
      UPDATE customers SET last_order = CURRENT_DATE() WHERE customer_id = 5001;
    END;
    """;

Statement stmt = conn.createStatement();
stmt.execute(sql);

Użycie z interfejsem API do wykonywania poleceń

Uruchom transakcje nieinterakcyjne przy użyciu interfejsu API wykonywania instrukcji:

import requests

sql = """
BEGIN ATOMIC
  INSERT INTO sales (sale_id, amount) VALUES (3001, 750.00);
  UPDATE daily_totals SET total = total + 750.00 WHERE sale_date = CURRENT_DATE();
END;
"""

response = requests.post(
    f"{workspace_url}/api/2.0/sql/statements",
    headers={"Authorization": f"Bearer {token}"},
    json={
        "warehouse_id": warehouse_id,
        "statement": sql,
        "wait_timeout": "30s"
    }
)

Wzorce ETL

Poniższe wzorce przedstawiają typowe przepływy pracy ETL korzystające z transakcji nieinterakcyjnych.

Wzorzec etapowania i walidacji

Ten wzorzec ładuje dane do obszaru przejściowego, weryfikuje jakość danych i scala zweryfikowane rekordy z tabelami produkcyjnymi:

BEGIN ATOMIC
  -- Load into staging
  INSERT INTO staging_customers
  SELECT * FROM external_source
  WHERE ingest_date = current_date();

  -- Validate data quality
  IF (SELECT COUNT(*) FROM staging_customers WHERE email NOT LIKE '%@%') > 0 THEN
    SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'Invalid email addresses found';
  END IF;

  -- Merge validated data
  MERGE INTO customers AS target
  USING staging_customers AS source
  ON target.customer_id = source.customer_id
  WHEN MATCHED THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *;

  -- Update metadata
  UPDATE etl_metadata
  SET last_load_date = current_date(),
      rows_processed = (SELECT COUNT(*) FROM staging_customers)
  WHERE table_name = 'customers';
END;

Wzorzec tabel wymiarów i faktów

Ten wzorzec aktualizuje tabele wymiarów przed załadowaniem tabel faktów w celu zachowania integralności referencyjnej:

BEGIN ATOMIC
  -- Update dimension tables first
  MERGE INTO dim_products AS target
  USING staging_products AS source
  ON target.product_id = source.product_id
  WHEN MATCHED THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *;

  MERGE INTO dim_customers AS target
  USING staging_customers AS source
  ON target.customer_id = source.customer_id
  WHEN MATCHED THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *;

  -- Then load fact table with foreign key references
  INSERT INTO fact_sales
  SELECT s.sale_id, p.product_id, c.customer_id, s.sale_amount, s.sale_date
  FROM staging_sales s
  JOIN dim_products p ON s.product_id = p.product_id
  JOIN dim_customers c ON s.customer_id = c.customer_id;
END;

Obsługa błędów

Gdy instrukcja kończy się niepowodzeniem w bloku BEGIN ATOMIC ... END;, Azure Databricks wycofa wszystkie zmiany i zwraca komunikat o błędzie.

Porady dotyczące debugowania:

1. Przejrzyj komunikat o błędzie, aby określić, która instrukcja nie powiodła się.
2. Testuj instrukcje indywidualnie poza blokiem transakcji.
3. Dodaj sprawdzanie poprawności przy użyciu polecenia SIGNAL , aby zakończyć się niepowodzeniem z niestandardowymi komunikatami o błędach.
4. Wykonywanie zapytań dotyczących historii transakcji w celu uzyskania dodatkowego kontekstu.

Transakcje interakcyjne

Transakcje interakcyjne zapewniają jawną kontrolę nad granicami transakcji. Należy ręcznie rozpocząć transakcję, uruchomić instrukcje i jawnie zatwierdzić lub wycofać.

Obsługiwane zasoby obliczeniowe: tylko magazyny SQL.

Obsługiwana składnia: tylko sql.

Składnia

Aby zatwierdzić:

BEGIN TRANSACTION;
statement1;
statement2;
COMMIT;

Aby przywrócić poprzednią wersję:

BEGIN TRANSACTION;
statement1;
statement2;
ROLLBACK;

Zweryfikuj poprawność przed zatwierdzeniem

Użyj transakcji interakcyjnych, aby zweryfikować wyniki przed zatwierdzeniem:

BEGIN TRANSACTION;

-- Load staging data
INSERT INTO staging_customers
SELECT * FROM external_customers
WHERE load_date = current_date();

-- Validate and commit or rollback
BEGIN
  DECLARE duplicate_count INT;
  SET duplicate_count = (
    SELECT COUNT(*) FROM (
      SELECT customer_id, COUNT(*) as cnt
      FROM staging_customers
      WHERE load_date = current_date()
      GROUP BY customer_id
      HAVING COUNT(*) > 1
    )
  );

  IF duplicate_count > 0 THEN
    ROLLBACK;
    SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'Duplicate customers found in staging data';
  ELSE
    MERGE INTO customers AS target
    USING staging_customers AS source
    ON target.customer_id = source.customer_id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *;
    COMMIT;
  END IF;
END;

Jawne wycofywanie

Wycofywanie transakcji w przypadku niepowodzenia walidacji lub logika biznesowa wymaga odrzucenia zmian:

BEGIN TRANSACTION;

UPDATE inventory
SET quantity = quantity - 50
WHERE product_id = 2001;

-- Check if quantity would go negative
BEGIN
  DECLARE new_quantity INT;
  SET new_quantity = (SELECT quantity FROM inventory WHERE product_id = 2001);

  IF new_quantity < 0 THEN
    ROLLBACK;
    SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'Insufficient inventory for product 2001';
  ELSE
    COMMIT;
  END IF;
END;

Używanie z funkcją JDBC

Sterownik JDBC obsługuje uruchamianie instrukcji DML przy użyciu executeUpdate() w ramach transakcji. Aby uzyskać listę obsługiwanych instrukcji DML, zobacz Obsługiwane operacje.

Klienci JDBC używają transakcji interakcyjnych, wyłączając tryb automatycznego zatwierdzania:

Connection conn = DriverManager.getConnection(jdbcUrl, properties);

try {
    conn.setAutoCommit(false);  // Start transaction mode
    Statement stmt = conn.createStatement();

    stmt.executeUpdate("INSERT INTO accounts (account_id, balance) VALUES (1001, 5000)");
    stmt.executeUpdate("UPDATE accounts SET balance = balance - 100 WHERE account_id = 1001");

    conn.commit();  // Commit the transaction

} catch (SQLException e) {
    conn.rollback();  // Roll back on error
    throw e;
} finally {
    conn.close();
}

Nieobsługiwane operacje JDBC

Następujące operacje JDBC nie są obsługiwane w ramach transakcji interakcyjnych:

Używanie z ODBC

Sterownik ODBC obsługuje wykonywanie instrukcji DML przy użyciu SQLExecute() i SQLExecDirect() w ramach transakcji. Aby uzyskać listę obsługiwanych instrukcji DML, zobacz Obsługiwane operacje.

Klienci ODBC mogą używać transakcji interakcyjnych ze sterownikiem Azure Databricks ODBC przy użyciu standardowych funkcji zarządzania transakcjami ODBC.

Nieobsługiwane operacje ODBC

Następujące operacje ODBC nie są obsługiwane w ramach transakcji interakcyjnych:

Użyj z łącznikiem SQL Databricks dla Pythona

Łącznik SQL usługi Databricks dla Python obsługuje uruchamianie instrukcji DML przy użyciu cursor.execute() w ramach transakcji. Aby uzyskać listę obsługiwanych instrukcji DML, zobacz Obsługiwane operacje.

Aplikacje Python mogą używać interaktywnych transakcji z łącznikiem Databricks SQL dla Pythona, ustawiając :

from databricks import sql

with sql.connect(
    server_hostname="dbc-a1b2345c-d6e7.cloud.databricks.com",
    http_path="sql/1.0/warehouses/abc123def456",
    access_token="your-access-token",
    autocommit=False
) as connection:
    with connection.cursor() as cursor:
        cursor.execute("INSERT INTO accounts (account_id, balance) VALUES (1001, 5000)")
        cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE account_id = 1001")
        connection.commit()

Nieobsługiwane operacje łącznika Python

Następujące operacje łącznika Python nie są obsługiwane w ramach transakcji interakcyjnych:

Ograniczenia sterowników dotyczące transakcji interakcyjnych

Następujące ograniczenia dotyczą wszystkich sterowników podczas korzystania z transakcji interaktywnych.

Operacje metadanych nie są obsługiwane wewnątrz transakcji interaktywnych. Następujące operacje mogą zakończyć się niepowodzeniem w ramach transakcji niezależnie od sterownika lub protokołu:

Usługa Databricks zaleca uruchamianie tych operacji metadanych poza transakcjami.

Zachowanie izolacji

Niezatwierdzone zmiany w transakcji interakcyjnej są widoczne tylko dla sesji. Zmiany są widoczne tylko dla innych sesji po uruchomieniu polecenia COMMIT;.

1. Aby zweryfikować izolację, utwórz przykładową tabelę, jeśli nie istnieje:

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

1. W tej samej sesji uruchom transakcję i wprowadź zmianę:

   BEGIN TRANSACTION;
   INSERT INTO sample_accounts VALUES (10, 'Test', 100.00);
   

2. W osobnej karcie edytora SQL lub sesji notesu (a nie nowej komórki w tym samym notesie) wykonaj zapytanie dotyczące tabeli:

   -- Run this in the SECOND session
   SELECT * FROM sample_accounts WHERE id = 10;
   

   Spowoduje to zwrócenie 0 wierszy, ponieważ niezatwierdzona zmiana nie jest widoczna poza Twoją pierwszą sesją.

3. Wróć do pierwszej sesji i zatwierdź:

   COMMIT;
   

4. Ponownie wykonaj zapytanie z drugiej sesji:

   -- Run this in the SECOND session
   SELECT * FROM sample_accounts WHERE id = 10;
   

   Wiersz jest widoczny, ponieważ transakcja została zatwierdzona.

Ta izolacja uniemożliwia innym użytkownikom odczytywanie danych, które mogą zostać wycofane.

Wybieranie trybu transakcji

Dodatkowe zasoby

• Samouczek: koordynowanie transakcji między tabelami
• Transakcje
• Zatwierdzenia wykazu
• Poziomy izolacji i konflikty zapisu
• INSTRUKCJE SKŁADNIOWE ATOMIC (transakcje bez interakcji): Uruchamiaj wiele instrukcji SQL jako pojedynczą transakcję atomową z automatycznym zatwierdzeniem i wycofywaniem.
• BEGIN TRANSACTION (transakcje interakcyjne): Rozpocznij interakcyjną transakcję z ręczną kontrolą zatwierdzania i wycofywania.
• ZATWIERDZENIE: zatwierdź transakcję interaktywną i wprowadź wszystkie zmiany trwałe.
• WYCOFYWANIE: Wycofywanie transakcji interakcyjnej i odrzucanie wszystkich zmian.