Dokumentacja języka Python w tabelach delta Live Tables

Ten artykuł zawiera szczegółowe informacje dotyczące interfejsu programowania Delta Live Tables w języku Python.

Aby uzyskać informacje na temat interfejsu API SQL, zobacz dokumentację języka SQL delta Live Tables.

Aby uzyskać szczegółowe informacje dotyczące konfigurowania automatycznego modułu ładującego, zobacz Co to jest moduł automatycznego ładowania?.

Przed rozpoczęciem

Poniżej przedstawiono ważne zagadnienia dotyczące implementowania potoków za pomocą interfejsu języka Python delta Live Tables:

• Ponieważ język Python table() i view() funkcje są wywoływane wielokrotnie podczas planowania i uruchamiania aktualizacji potoku, nie dołączaj kodu do jednej z tych funkcji, które mogą mieć skutki uboczne (na przykład kod modyfikujący dane lub wysyłający wiadomość e-mail). Aby uniknąć nieoczekiwanego zachowania, funkcje języka Python definiujące zestawy danych powinny zawierać tylko kod wymagany do zdefiniowania tabeli lub widoku.
• Aby wykonywać operacje, takie jak wysyłanie wiadomości e-mail lub integrowanie z zewnętrzną usługą monitorowania, szczególnie w funkcjach definiujących zestawy danych, użyj punktów zaczepienia zdarzeń. Zaimplementowanie tych operacji w funkcjach definiujących zestawy danych spowoduje nieoczekiwane zachowanie.
• Język Python table i view funkcje muszą zwracać ramkę danych. Niektóre funkcje działające na ramkach danych nie zwracają ramek danych i nie powinny być używane. Te operacje obejmują funkcje, takie jak collect(), , count()toPandas(), save()i saveAsTable(). Ponieważ przekształcenia ramki danych są wykonywane po rozwiązaniu pełnego grafu przepływu danych, użycie takich operacji może mieć niezamierzone skutki uboczne.

Importowanie modułu dlt języka Python

Funkcje języka Python w tabelach na żywo funkcji delta są definiowane w dlt module. Potoki zaimplementowane za pomocą interfejsu API języka Python muszą zaimportować ten moduł:

import dlt

Tworzenie tabel delta live zmaterializowanego widoku lub tabeli przesyłania strumieniowego

W języku Python funkcja Delta Live Tables określa, czy zestaw danych ma być aktualizowany jako zmaterializowany widok, czy tabela przesyłania strumieniowego na podstawie definiującego zapytania. Dekorator @table może służyć do definiowania zarówno zmaterializowanych widoków, jak i tabel przesyłania strumieniowego.

Aby zdefiniować zmaterializowany widok w języku Python, zastosuj do @table zapytania, które wykonuje statyczny odczyt względem źródła danych. Aby zdefiniować tabelę przesyłania strumieniowego, zastosuj się @table do zapytania, które wykonuje odczyt strumieniowy względem źródła danych lub użyj funkcji create_streaming_table(). Oba typy zestawów danych mają tę samą specyfikację składni w następujący sposób:

import dlt

@dlt.table(
  name="<name>",
  comment="<comment>",
  spark_conf={"<key>" : "<value>", "<key>" : "<value>"},
  table_properties={"<key>" : "<value>", "<key>" : "<value>"},
  path="<storage-location-path>",
  partition_cols=["<partition-column>", "<partition-column>"],
  cluster_by = ["<clustering-column>", "<clustering-column>"],
  schema="schema-definition",
  row_filter = "row-filter-clause",
  temporary=False)
@dlt.expect
@dlt.expect_or_fail
@dlt.expect_or_drop
@dlt.expect_all
@dlt.expect_all_or_drop
@dlt.expect_all_or_fail
def <function-name>():
    return (<query>)

Tworzenie widoku tabel na żywo delty

Aby zdefiniować widok w języku Python, zastosuj @view dekorator. @table Podobnie jak dekorator, można użyć widoków w tabelach delta live dla zestawów danych statycznych lub przesyłanych strumieniowo. Poniżej przedstawiono składnię definiującą widoki przy użyciu języka Python:

import dlt

@dlt.view(
  name="<name>",
  comment="<comment>")
@dlt.expect
@dlt.expect_or_fail
@dlt.expect_or_drop
@dlt.expect_all
@dlt.expect_all_or_drop
@dlt.expect_all_or_fail
def <function-name>():
    return (<query>)

Przykład: Definiowanie tabel i widoków

Aby zdefiniować tabelę lub widok w języku Python, zastosuj @dlt.view funkcję lub @dlt.table dekorator. Możesz użyć nazwy funkcji lub parametru name , aby przypisać tabelę lub nazwę widoku. W poniższym przykładzie zdefiniowano dwa różne zestawy danych: widok o nazwie taxi_raw , który przyjmuje plik JSON jako źródło wejściowe i tabelę o nazwie filtered_data , która przyjmuje taxi_raw widok jako dane wejściowe:

import dlt

@dlt.view
def taxi_raw():
  return spark.read.format("json").load("/databricks-datasets/nyctaxi/sample/json/")

# Use the function name as the table name
@dlt.table
def filtered_data():
  return dlt.read("taxi_raw").where(...)

# Use the name parameter as the table name
@dlt.table(
  name="filtered_data")
def create_filtered_data():
  return dlt.read("taxi_raw").where(...)

Przykład: uzyskiwanie dostępu do zestawu danych zdefiniowanego w tym samym potoku

Oprócz odczytywania z zewnętrznych źródeł danych można uzyskać dostęp do zestawów danych zdefiniowanych w tym samym potoku za pomocą funkcji Delta Live Tables read() . W poniższym przykładzie pokazano tworzenie customers_filtered zestawu danych przy użyciu read() funkcji :

@dlt.table
def customers_raw():
  return spark.read.format("csv").load("/data/customers.csv")

@dlt.table
def customers_filteredA():
  return dlt.read("customers_raw").where(...)

Możesz również użyć spark.table() funkcji , aby uzyskać dostęp do zestawu danych zdefiniowanego w tym samym potoku. W przypadku uzyskiwania spark.table() dostępu do zestawu danych zdefiniowanego w potoku przy użyciu funkcji argument funkcji poprzedzał LIVE słowo kluczowe nazwą zestawu danych:

@dlt.table
def customers_raw():
  return spark.read.format("csv").load("/data/customers.csv")

@dlt.table
def customers_filteredB():
  return spark.table("LIVE.customers_raw").where(...)

Przykład: odczyt z tabeli zarejestrowanej w magazynie metadanych

Aby odczytać dane z tabeli zarejestrowanej w magazynie metadanych Hive, w argumencie funkcji pomiń LIVE słowo kluczowe i opcjonalnie zakwalifikuj nazwę tabeli o nazwie bazy danych:

@dlt.table
def customers():
  return spark.table("sales.customers").where(...)

Aby zapoznać się z przykładem odczytu z tabeli wykazu aparatu Unity, zobacz Pozyskiwanie danych do potoku wykazu aparatu Unity.

Przykład: uzyskiwanie dostępu do zestawu danych przy użyciu spark.sql

Zestaw danych można również zwrócić przy użyciu spark.sql wyrażenia w funkcji zapytania. Aby odczytać z wewnętrznego zestawu danych, należy LIVE. wstępnie otworzyć nazwę zestawu danych:

@dlt.table
def chicago_customers():
  return spark.sql("SELECT * FROM LIVE.customers_cleaned WHERE city = 'Chicago'")

Tworzenie tabeli, która ma być używana jako element docelowy operacji przesyłania strumieniowego

create_streaming_table() Użyj funkcji , aby utworzyć tabelę docelową dla rekordów wyjściowych przez operacje przesyłania strumieniowego, w tym apply_changes(), apply_changes_from_snapshot()i rekordy wyjściowe @append_flow.

create_streaming_table(
  name = "<table-name>",
  comment = "<comment>"
  spark_conf={"<key>" : "<value", "<key" : "<value>"},
  table_properties={"<key>" : "<value>", "<key>" : "<value>"},
  partition_cols=["<partition-column>", "<partition-column>"],
  cluster_by = ["<clustering-column>", "<clustering-column>"],
  path="<storage-location-path>",
  schema="schema-definition",
  expect_all = {"<key>" : "<value", "<key" : "<value>"},
  expect_all_or_drop = {"<key>" : "<value", "<key" : "<value>"},
  expect_all_or_fail = {"<key>" : "<value", "<key" : "<value>"},
  row_filter = "row-filter-clause"
)

Kontrolowanie sposobu materializacji tabel

Tabele oferują również dodatkową kontrolę nad ich materializacją:

• Określ sposób partycjonowania tabel przy użyciu polecenia partition_cols. Możesz użyć partycjonowania, aby przyspieszyć wykonywanie zapytań.
• Właściwości tabeli można ustawić podczas definiowania widoku lub tabeli. Zobacz Właściwości tabeli Tabele na żywo funkcji Delta.
• Ustaw lokalizację przechowywania danych tabeli przy użyciu path ustawienia . Domyślnie dane tabeli są przechowywane w lokalizacji magazynu potoku, jeśli path nie są ustawione.
• W definicji schematu można użyć wygenerowanych kolumn . Zobacz Przykład: określanie schematu i kolumn partycji.

Przykład: określanie schematu i kolumn partycji

Opcjonalnie można określić schemat tabeli przy użyciu języka Python StructType lub ciągu DDL SQL. Po określeniu ciągu DDL definicja może zawierać wygenerowane kolumny.

Poniższy przykład tworzy tabelę o nazwie sales ze schematem określonym przy użyciu języka Python StructType:

sales_schema = StructType([
  StructField("customer_id", StringType(), True),
  StructField("customer_name", StringType(), True),
  StructField("number_of_line_items", StringType(), True),
  StructField("order_datetime", StringType(), True),
  StructField("order_number", LongType(), True)]
)

@dlt.table(
  comment="Raw data on sales",
  schema=sales_schema)
def sales():
  return ("...")

Poniższy przykład określa schemat tabeli przy użyciu ciągu DDL, definiuje wygenerowaną kolumnę i definiuje kolumnę partycji:

@dlt.table(
  comment="Raw data on sales",
  schema="""
    customer_id STRING,
    customer_name STRING,
    number_of_line_items STRING,
    order_datetime STRING,
    order_number LONG,
    order_day_of_week STRING GENERATED ALWAYS AS (dayofweek(order_datetime))
    """,
  partition_cols = ["order_day_of_week"])
def sales():
  return ("...")

Domyślnie tabele na żywo delty wywnioskują schemat z table definicji, jeśli nie określisz schematu.

Konfigurowanie tabeli przesyłania strumieniowego w celu ignorowania zmian w źródłowej tabeli przesyłania strumieniowego

Domyślnie tabele przesyłania strumieniowego wymagają źródeł tylko do dołączania. Gdy tabela przesyłania strumieniowego używa innej tabeli przesyłania strumieniowego jako źródła, a źródłowa tabela przesyłania strumieniowego wymaga aktualizacji lub usunięcia, na przykład przetwarzania RODO "prawo do zapomnienia" skipChangeCommits można ustawić flagę podczas odczytywania źródłowej tabeli przesyłania strumieniowego w celu zignorowania tych zmian. Aby uzyskać więcej informacji na temat tej flagi, zobacz Ignorowanie aktualizacji i usuwania.

@table
def b():
   return spark.readStream.option("skipChangeCommits", "true").table("LIVE.A")

Przykład: Definiowanie ograniczeń tabeli

Podczas określania schematu można zdefiniować klucze podstawowe i obce. Ograniczenia są informacyjne i nie są wymuszane. Zobacz klauzulę CONSTRAINT w dokumentacji języka SQL.

W poniższym przykładzie zdefiniowano tabelę z ograniczeniem klucza podstawowego i obcego:

@dlt.table(
   schema="""
    customer_id STRING NOT NULL PRIMARY KEY,
    customer_name STRING,
    number_of_line_items STRING,
    order_datetime STRING,
    order_number LONG,
    order_day_of_week STRING GENERATED ALWAYS AS (dayofweek(order_datetime)),
    CONSTRAINT fk_customer_id FOREIGN KEY (customer_id) REFERENCES main.default.customers(customer_id)
    """
def sales():
   return ("...")

Przykład: Definiowanie filtru wierszy i maski kolumn

Aby utworzyć zmaterializowany widok lub tabelę strumieniową z filtrem wierszy i maską kolumny, użyj klauzuli ROW FILTER i klauzuli MASK. W poniższym przykładzie pokazano, jak zdefiniować zmaterializowany widok i tabelę przesyłania strumieniowego z filtrem wierszy i maską kolumn:

@dlt.table(
   schema="""
    id int COMMENT 'This is the customer ID',
    name string COMMENT 'This is the customer full name',
    region string,
    ssn string MASK catalog.schema.ssn_mask_fn USING COLUMNS (region)
    """,
  row_filter = "ROW FILTER catalog.schema.us_filter_fn ON (region, name)"
def sales():
   return ("...")

Aby uzyskać więcej informacji na temat filtrów wierszy i masek kolumn, zobacz Publikowanie tabel z filtrami wierszy i maskami kolumn.

Właściwości tabel delta live języka Python

W poniższych tabelach opisano opcje i właściwości, które można określić podczas definiowania tabel i widoków za pomocą tabel różnicowych na żywo:

Zmienianie przechwytywania danych z zestawienia zmian za pomocą języka Python w tabelach delta live tables

apply_changes() Użyj funkcji w interfejsie API języka Python, aby używać funkcji przechwytywania zmian danych w tabelach delta Live Tables (CDC) do przetwarzania danych źródłowych ze źródła danych zmian (CDF).

apply_changes(
  target = "<target-table>",
  source = "<data-source>",
  keys = ["key1", "key2", "keyN"],
  sequence_by = "<sequence-column>",
  ignore_null_updates = False,
  apply_as_deletes = None,
  apply_as_truncates = None,
  column_list = None,
  except_column_list = None,
  stored_as_scd_type = <type>,
  track_history_column_list = None,
  track_history_except_column_list = None
)

Aby dowiedzieć się więcej na temat przetwarzania CDC za pomocą zestawienia zmian, zobacz Interfejsy API ZASTOSUJ ZMIANY: Upraszczanie przechwytywania zmian przy użyciu tabel funkcji Delta Live Tables. Aby zapoznać się z przykładem użycia apply_changes() funkcji, zobacz Przykład: typ SCD 1 i typ SCD 2 przetwarzanie z danymi źródłowymi CDF.

Zobacz Interfejsy API ZASTOSUJ ZMIANY: upraszczanie przechwytywania danych zmian za pomocą tabel różnicowych na żywo.

Zmienianie przechwytywania danych z migawek bazy danych za pomocą języka Python w tabelach delta live

apply_changes_from_snapshot() Użyj funkcji w interfejsie API języka Python, aby używać funkcji przechwytywania zmian danych funkcji delta live tables (CDC) do przetwarzania danych źródłowych z migawek bazy danych.

apply_changes_from_snapshot(
  target = "<target-table>",
  source = Any,
  keys = ["key1", "key2", "keyN"],
  stored_as_scd_type = <type>,
  track_history_column_list = None,
  track_history_except_column_list = None
) -> None

Aby dowiedzieć się więcej na temat przetwarzania CDC za pomocą migawek, zobacz Interfejsy API ZASTOSUJ ZMIANY: Upraszczanie przechwytywania zmian przy użyciu tabel delta live. Przykłady użycia funkcji można znaleźć w przykładach okresowego pozyskiwania apply_changes_from_snapshot() migawek i historycznych przykładów pozyskiwania migawek.

Implementowanie argumentu source

Funkcja apply_changes_from_snapshot() zawiera source argument . W przypadku przetwarzania migawek historycznych argument powinien być funkcją lambda języka Python, source która zwraca dwie wartości do apply_changes_from_snapshot() funkcji: ramkę danych języka Python zawierającą dane migawki do przetworzenia i wersję migawki.

Poniżej znajduje się podpis funkcji lambda:

lambda Any => Optional[(DataFrame, Any)]

• Argument funkcji lambda jest ostatnio przetworzoną wersją migawki.
• Wartość zwracana funkcji lambda to None lub krotka dwóch wartości: Pierwsza wartość krotki to ramka danych zawierająca migawkę do przetworzenia. Drugą wartością krotki jest wersja migawki, która reprezentuje kolejność logiczną migawki.

Przykład, który implementuje i wywołuje funkcję lambda:

def next_snapshot_and_version(latest_snapshot_version):
 if latest_snapshot_version is None:
   return (spark.read.load("filename.csv"), 1)
 else:
   return None

apply_changes_from_snapshot(
  # ...
  source = next_snapshot_and_version,
  # ...
)

Środowisko uruchomieniowe delta Live Tables wykonuje następujące kroki za każdym razem, gdy potok zawierający apply_changes_from_snapshot() funkcję jest wyzwalany:

1. Uruchamia funkcję w next_snapshot_and_version celu załadowania następnej ramki danych migawki i odpowiedniej wersji migawki.
2. Jeśli żadna ramka danych nie zostanie zwrócona, przebieg zostanie zakończony, a aktualizacja potoku zostanie oznaczona jako ukończona.
3. Wykrywa zmiany w nowej migawki i przyrostowo stosuje je do tabeli docelowej.
4. Wraca do kroku 1, aby załadować następną migawkę i jego wersję.

Ograniczenia

Interfejs delta Live Tables w języku Python ma następujące ograniczenie:

Funkcja nie jest obsługiwana pivot() . Operacja pivot na platformie Spark wymaga chętnego ładowania danych wejściowych w celu obliczenia schematu danych wyjściowych. Ta funkcja nie jest obsługiwana w tabelach delta live.