Referenční dokumentace jazyka PYTHONu DLT

Tento článek obsahuje podrobnosti o programovacím rozhraní DLT Python.

Informace o rozhraní SQL API najdete v referenční jazyka SQLDLT .

Podrobnosti týkající se konfigurace automatického zavaděče najdete v tématu Co je automatický zavaděč?.

Než začnete

Při implementaci kanálů s rozhraním DLT Python je důležité vzít v úvahu následující skutečnosti:

• Vzhledem k tomu, že se funkce pythonu table() a view() vyvolávají několikrát během plánování a spuštění aktualizace kanálu, nezahrnujte kód do jedné z těchto funkcí, která může mít vedlejší účinky (například kód, který upravuje data nebo odesílá e-mail). Aby nedocházelo k neočekávanému chování, měly by funkce Pythonu, které definují datové sady, obsahovat pouze kód potřebný k definování tabulky nebo zobrazení.
• K provádění operací, jako je odesílání e-mailů nebo integrace s externí monitorovací službou, zejména ve funkcích, které definují datové sady, použijte událostní háčky. Implementace těchto operací ve funkcích, které definují datové sady, způsobí neočekávané chování.
• Funkce table a view Pythonu musí vracet datový rámec. Některé funkce, které pracují s datovými rámci, nevrací datové rámce a neměly by se používat. Mezi tyto operace patří funkce, jako jsou collect(), count(), toPandas(), save()a saveAsTable(). Vzhledem k tomu, že transformace datového rámce se provádějí po vyřešení úplného grafu toku dat, může použití těchto operací mít nežádoucí vedlejší účinky.

Import modulu dlt Pythonu

Funkce DLT Pythonu jsou definovány v modulu dlt. Vaše kanály implementované pomocí rozhraní Python API musí importovat tento modul:

import dlt

Vytvoření materializovaného zobrazení DLT nebo tabulky streamování

DlT v Pythonu určuje, jestli se má datová sada aktualizovat jako materializovaná nebo streamovaná tabulka na základě definovaného dotazu. Dekorátor @table lze použít k definování materializovaných zobrazení i streamovaných tabulek.

Pokud chcete definovat materializované zobrazení v Pythonu, použijte @table na dotaz, který provádí statické čtení proti zdroji dat. Pokud chcete definovat streamovací tabulku, použijte @table na dotaz, který provádí streamování čtení proti zdroji dat, nebo použijte funkci create_streaming_table(). Oba typy datových sad mají stejnou specifikaci syntaxe:

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

Vytvoření zobrazení DLT

Pokud chcete definovat zobrazení v Pythonu, použijte @view dekorátor. Podobně jako dekorátor @table můžete v DLT použít pohledy pro statické nebo streamované datové sady. Následuje syntaxe pro definování zobrazení pomocí Pythonu:

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

Příklad: Definování tabulek a zobrazení

Pokud chcete definovat tabulku nebo zobrazení v Pythonu, použijte pro funkci @dlt.view nebo @dlt.table dekorátor. Název funkce nebo parametr name můžete použít k přiřazení názvu tabulky nebo zobrazení. Následující příklad definuje dvě různé datové sady: zobrazení označované jako taxi_raw, které jako vstupní zdroj přebírá soubor JSON a tabulku s názvem filtered_data, která přebírá taxi_raw zobrazení jako vstup:

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 spark.read.table("taxi_raw").where(...)

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

Příklad: Přístup k datové sadě definované ve stejném kanálu

Pokud chcete získat přístup k datové sadě definované ve stejném kanálu, použijte funkce spark.read.table() nebo spark.readStream.table():

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

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

Příklad: Čtení z tabulky zaregistrované v metastoru

Pokud chcete číst data z tabulky zaregistrované v metastoru Hive, můžete v argumentu funkce kvalifikovat název tabulky s názvem databáze:

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

Příklad čtení z tabulky Katalogu Unity najdete v tématu Ingestování dat do kanálu katalogu Unity.

Příklad : Přístup k datové sadě pomocí spark.sql

Datovou sadu můžete vrátit také pomocí výrazu spark.sql ve funkci dotazu. Pokud chcete číst z interní datové sady, můžete buď ponechat název nekvalifikovaný pro použití výchozího katalogu a schématu, nebo je můžete předem nastavit:

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

Trvalé odstranění záznamů z materializovaného zobrazení nebo tabulky streamování

Pokud chcete trvale odstranit záznamy z materializovaného zobrazení nebo tabulky streamování s povolenými vektory odstranění, jako je například dodržování předpisů GDPR, musí být u podkladových tabulek Delta objektu provedeny další operace. Pokud chcete zajistit odstranění záznamů z materializovaného zobrazení, přečtěte si Trvalé odstranění záznamů z materializovaného zobrazení s povolenými vektory odstranění. Pokud chcete zajistit odstranění záznamů z tabulky streamování, přečtěte si téma Trvalé odstranění záznamů z tabulky streamování.

Zápis do externích služeb streamování událostí nebo tabulek Delta pomocí rozhraní DLT sink API

Pokud chcete zapisovat do služby streamování událostí, jako je Apache Kafka nebo Azure Event Hubs nebo do tabulky Delta z kanálu DLT, použijte funkci create_sink(), která je součástí modulu dlt Pythonu. Po vytvoření jímky pomocí funkce create_sink() použijete jímku v připojit tok k zápisu dat do jímky. Tok "Append" je jediný typ toku podporovaný funkcí create_sink(). Jiné typy toků, například apply_changes, se nepodporují.

Následuje syntaxe pro vytvoření jímky s funkcí create_sink():

create_sink(<sink_name>, <format>, <options>)

příklad : Vytvoření jímky Kafka s funkcí create_sink()

create_sink(
  "my_kafka_sink",
  "kafka",
  {
    "kafka.bootstrap.servers": "host:port",
    "topic": "my_topic"
  }
)

příklad : Vytvoření jímky Delta s funkcí create_sink() a cestou k systému souborů

Následující příklad vytvoří zdroj, který zapisuje do tabulky Delta tím, že zadá cestu filesystému k tabulce:

create_sink(
  "my_delta_sink",
    "delta",
    { "path": "//path/to/my/delta/table" }
)

příklad : Vytvoření jímky Delta s funkcí create_sink() a názvem tabulky Katalogu Unity

Následující příklad vytvoří jímku, která zapisuje do tabulky Delta předáním názvu tabulky v Katalogu Unity:

create_sink(
  "my_delta_sink",
    "delta",
    { "tableName": "my_catalog.my_schema.my_table" }
)

Příklad: Použití zapisovacího toku k zápisu do úložiště Delta

Následující příklad vytvoří jímku, která zapíše do tabulky Delta a pak vytvoří tok připojení pro zápis do této jímky:

create_sink("my_sink", "delta", {"path": "/tmp/delta_sink"})

@append_flow(name = "flow", target = "my_sink")
def flowFunc():
  return <streaming-query>

příklad : Použití toku připojení k zápisu do jímky Kafka

Následující příklad vytvoří jímku, která zapíše do tématu Kafka a pak vytvoří tok připojení pro zápis do této jímky:

create_sink(
  "my_kafka_sink",
  "kafka",
  {
    "kafka.bootstrap.servers": "host:port",
    "topic": "my_topic"
  }
)

@append_flow(name = "flow", target = "my_kafka_sink")
def myFlow():
  return read_stream("xxx").select(F.to_json(F.struct("*")).alias("value"))

Schéma datového rámce zapsaného do systému Kafka by mělo obsahovat sloupce zadané v Konfigurace zapisovače strukturovaného streamování Kafka.

Vytvoření tabulky, která se použije jako cíl operací streamování

Pomocí funkce create_streaming_table() vytvořte cílovou tabulku pro záznamy, které jsou výstupem streamovacích operací, včetně apply_changes(), apply_changes_from_snapshot()a @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"
)

Řízení způsobu materializace tabulek

Tabulky také nabízejí další kontrolu nad jejich materializací:

• Určete, jak tabulky clusteru pomocí cluster_by. Můžete použít liquid clustering k urychlení dotazů. Viz Použijte kapalinové seskupování pro tabulky Delta.
• Určete, jak jsou tabulky rozdělovány pomocí partition_cols.
• Vlastnosti tabulky můžete nastavit při definování zobrazení nebo tabulky. Viz vlastnosti tabulky DLT.
• Pomocí nastavení path nastavte umístění úložiště pro data tabulky. Ve výchozím nastavení se data tabulky ukládají do umístění úložiště kanálu, pokud path není nastaveno.
• V definici schématu můžete použít sloupce vygenerované . Viz Příklad: Určení schématu a sloupců clusteru.

příklad : Zadání schématu a sloupců clusteru

Volitelně můžete zadat schéma tabulky pomocí pythonového StructType nebo řetězce SQL DDL. Pokud je specifikováno pomocí DDL řetězce, může definice zahrnovat generované sloupce.

Následující příklad vytvoří tabulku s názvem sales se schématem zadaným pomocí StructTypePythonu:

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 ("...")

Následující příklad určuje schéma tabulky pomocí řetězce DDL, definuje vygenerovaný sloupec a definuje sloupce clusteringu:

@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))
    """,
  cluster_by = ["order_day_of_week", "customer_id"])
def sales():
  return ("...")

Ve výchozím nastavení DLT odvodí schéma z definice table, pokud nezadáte schéma.

Příklad : Specifikovat sloupce oddílů

Následující příklad určuje schéma tabulky pomocí řetězce DDL, definuje vygenerovaný sloupec a definuje partiční sloupec:

@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 ("...")

Příklad: Definování omezení tabulky

Při zadávání schématu můžete definovat primární a cizí klíče. Omezení jsou informativní a nevynucují se. Vizte CONSTRAINT klauzuli v referenční dokumentaci jazyka SQL.

Následující příklad definuje tabulku s omezením primárního a cizího klíče:

@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 ("...")

Příklad: Definování filtru řádků a masky sloupců

Pokud chcete vytvořit materializované zobrazení nebo tabulku Streamování s filtrem řádků a maskou sloupců, použijte klauzuli ROW FILTER a klauzuli MASK. Následující příklad ukazuje, jak definovat materializované zobrazení a streamovací tabulku s filtrem řádků i maskou sloupce:

@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 ("...")

Další informace o filtrech řádků a maskách sloupců najdete v tématu Publikování tabulek s filtry řádků a maskami sloupců.

Konfigurace streamované tabulky tak, aby ignorovala změny ve zdrojové streamovací tabulce

Ve výchozím nastavení streamované tabulky vyžadují zdroje, které umožňují pouze přidávání. Pokud streamovací tabulka používá jako zdroj jinou streamovací tabulku a zdrojová streamovací tabulka vyžaduje aktualizace nebo odstranění, například zpracování gdpr "právo na zapomenutí", můžete při čtení zdrojové tabulky streamování nastavit příznak skipChangeCommits, aby se tyto změny ignorovaly. Další informace o tomto příznaku naleznete v tématu Ignorovat aktualizace a odstranit.

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

vlastnosti Python DLT

Následující tabulky popisují možnosti a vlastnosti, které můžete zadat při definování tabulek a zobrazení pomocí DLT:

Zachytávání dat změn z kanálu změn pomocí Pythonu v DLT

Pomocí funkce apply_changes() v rozhraní Python API můžete ke zpracování zdrojových dat z datového kanálu změn (CDF) použít funkci zachytávání dat DLT (CDC).

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
)

Další informace o zpracování CDC pomocí kanálu změn najdete v tématu Rozhraní API APPLY CHANGES: Zjednodušení zachytávání dat změn pomocí knihovny DLT. Pro příklad použití funkce apply_changes(), viz Příklad: Zpracování SCD typu 1 a typu 2 se zdrojovými daty CDF.

Viz rozhraní API APPLY CHANGES : Zjednodušte zachytávání dat změn pomocí DLT.

Změna zachytávání dat ze snímků databáze pomocí Pythonu v DLT

Pomocí funkce apply_changes_from_snapshot() v rozhraní Python API můžete ke zpracování zdrojových dat ze snímků databáze použít funkci zachytávání dat změn DLT (CDC).

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

Další informace o zpracování CDC pomocí snímků najdete v části s rozhraními API "APPLY CHANGES": Zjednodušte snímání změn dat pomocí DLT (a). Příklady použití funkce apply_changes_from_snapshot() najdete v příkladech periodického příjmu snímků a historického příjmu snímků.

Implementovat argument source

Funkce apply_changes_from_snapshot() obsahuje argument source. U zpracování historických snímků se očekává, že argument source bude funkcí lambda Pythonu, která vrátí dvě hodnoty do funkce apply_changes_from_snapshot(): datový rámec Pythonu obsahující data snímku, která se mají zpracovat, a verzi snímku.

Následuje podpis funkce lambda:

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

• Argumentem funkce lambda je nejnovější zpracovaná verze snímku.
• Návratová hodnota funkce lambda je None nebo řazená kolekce členů dvou hodnot: První hodnota řazené kolekce členů je datový rámec obsahující snímek, který se má zpracovat. Druhá hodnota n-tice je verze snímku, která představuje logické pořadí snímku.

Příklad, který implementuje a volá funkci 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,
  # ...
)

Modul runtime DLT provede následující kroky pokaždé, když se aktivuje kanál obsahující funkci apply_changes_from_snapshot():

1. Spustí funkci next_snapshot_and_version, která načte další datový rámec snímku a odpovídající verzi snímku.
2. Pokud se žádný datový rámec nevrátí, spuštění se ukončí a aktualizace datového toku se označí jako dokončená.
3. Rozpozná změny v novém snímku a postupně je použije na cílovou tabulku.
4. Vrátí se ke kroku 1, aby se načetl další snímek a jeho verze.

omezení

Rozhraní DLT Python má následující omezení:

Funkce pivot() není podporována. Operace pivot ve Sparku vyžaduje dychtivé načítání vstupních dat pro výpočet výstupního schématu. Tato funkce není v DLT podporována.