Ingest dat do Unity Catalogu

Než se pustíme do jednotlivých technik, podívejme se na celkovou mapu. Do Unity Catalogu vede několik ingesčních cest a každá má svůj sweet spot: managed konektory pro SaaS a databáze, notebook s vlastním kódem pro cokoli nestandardního, SQL příkazy pro jednorázové i opakované načítání souborů, Auto Loader pro průběžně přibývající soubory a Structured Streaming pro message busy. Všechny cesty končí ve stejném bodě — v Delta tabulkách bronze vrstvy spravovaných Unity Catalogem.

Lakeflow Connect je sada managed konektorů — místo psaní extrakčního kódu jen nakonfigurujete pipeline — přes UI, Asset Bundles (dnes Declarative Automation Bundles, viz sekce 12.4), notebook nebo CLI — a Databricks se stará o extrakci, staging i zápis do streaming tabulek. Konektory se dělí do tří kategorií:

Každá ingesční pipeline se skládá ze tří komponent: connection (uložené přihlašovací údaje a endpoint — vytvoříte jednou, použijete opakovaně), ingestion definition (které tabulky a kam v Unity Catalogu) a schedule. U databázových konektorů vzniká navíc ingestion gateway, která průběžně vytahuje změny ze zdrojové databáze.

# ingestion definition v Declarative Automation Bundles (drive Asset Bundles)
resources:
  pipelines:
    pipeline_sqlserver:
      name: customer-orders-pipeline
      catalog: sales
      schema: bronze
      ingestion_definition:
        # connection_name odkazuje na UC connection se zdrojovymi udaji
        connection_name: sqlserver_prod_connection
        # u databazovych konektoru navic gateway pipeline se zmenami ze zdroje
        ingestion_gateway_id: ${resources.pipelines.gateway_sqlserver.id}
        objects:
          - table:
              source_schema: dbo
              source_table: orders
              destination_catalog: sales
              destination_schema: bronze

Klíčové volby při konfiguraci: SCD type 1 (cíl vždy odráží aktuální stav zdroje — update přepíše řádek) vs. SCD type 2 (historie se uchovává, tabulka dostane sloupce __START_AT a __END_AT). Dále výběr sloupců přes include_columns / exclude_columns a možnost full refresh, když se inkrementální zpracování nedokáže zotavit ze změny schématu.

Když managed konektor neexistuje nebo potřebujete vlastní logiku, sáhnete po notebooku. Základem batch ingestu je spark.read — vrací DataFrame, distribuovanou kolekci dat organizovanou do pojmenovaných sloupců.

df = (spark.read
    .format("csv")
    .option("header", "true")
    .option("inferSchema", "true")
    .load("/Volumes/catalog/schema/volume/data.csv"))

# zapis do Delta tabulky v Unity Catalogu
df.write.mode("overwrite").saveAsTable("catalog.schema.table_name")

Stejným vzorem načtete JSON, Parquet i XML (tam je povinná option rowTag, která říká, který XML element odpovídá jednomu řádku). Zápis vždy končí v Delta formátu — ten je v Databricks výchozí a dává ACID transakce a time travel. Chování při existující tabulce řídí mode():

Pro relační zdroje bez Lakeflow konektoru poslouží JDBC. Přihlašovací údaje nikdy nepište do notebooku — patří do Databricks secrets:

df_jdbc = (spark.read
    .format("jdbc")
    .option("url", "jdbc:sqlserver://server.database.windows.net:1433;database=mydb")
    .option("dbtable", "schema.tablename")
    .option("user", dbutils.secrets.get(scope="jdbc", key="username"))
    .option("password", dbutils.secrets.get(scope="jdbc", key="password"))
    .load())

from pyspark.sql.functions import parse_json, variant_get, col

df = spark.read.text("/path/to/data.json")
df_variant = df.select(parse_json(col("value")).alias("data"))
df_variant.select(variant_get(col("data"), "$.customer.name", "string")).display()

Kdo preferuje deklarativní SQL před procedurálním kódem, má k dispozici tři příkazy. Liší se hlavně tím, co udělají s existující tabulkou — a to je přesně to, co si u nich musíte zapamatovat.

CTAS (CREATE TABLE AS SELECT) spojí vytvoření tabulky a naplnění daty do jednoho kroku. Schéma se odvodí z výsledku dotazu. Pro čtení ze souborů se kombinuje s table-valued funkcí read_files:

CREATE TABLE catalog.schema.sales_data AS
SELECT * FROM read_files(
    '/Volumes/catalog/schema/volume/sales/*.parquet',
    format => 'parquet'
);

CREATE OR REPLACE TABLE tabulku kompletně nahradí, ale na rozdíl od DROP + CREATE zachová historii tabulky, udělená oprávnění, row filters i column masks. Downstream konzumenti nic nepoznají. Hodí se na periodické full refreshe.

COPY INTO je určeno pro opakované přiléhání souborů: načte soubory z cloud storage a appenduje je do existující tabulky. Jeho klíčová vlastnost je idempotence — už jednou načtené soubory automaticky přeskočí, i když příkaz spustíte znovu:

-- cilova tabulka musi existovat predem
CREATE TABLE IF NOT EXISTS catalog.schema.events (
    event_id STRING, event_type STRING,
    event_timestamp TIMESTAMP, payload STRING);

COPY INTO catalog.schema.events
FROM '/Volumes/catalog/schema/volume/events/'
FILEFORMAT = JSON
FORMAT_OPTIONS ('multiline' = 'true');

Výběr souborů zúžíte přes PATTERN (glob) nebo FILES (konkrétní jména). Volba COPY_OPTIONS ('mergeSchema' = 'true') dovolí schématu tabulky evolvovat, a VALIDATE ALL zkontroluje parsovatelnost a shodu se schématem, aniž by cokoliv načetlo.

Když zdrojový systém generuje miliony transakcí denně, plné reloady přestávají dávat smysl. Change data capture (CDC) přenáší jen změny — každý záznam feedu nese datové sloupce, typ operace (INSERT/UPDATE/DELETE) a sekvenční sloupec (timestamp nebo pořadové číslo), který určuje pořadí změn.

Ruční aplikace CDC feedu je plná záludností: out-of-order události, deduplikace, mazání. AUTO CDC API v Lakeflow Spark Declarative Pipelines tohle řeší za vás — vytvoříte cílovou streaming tabulku, definujete flow a zvolíte SCD typ:

CREATE OR REFRESH STREAMING TABLE employees;

CREATE FLOW employees_cdc_flow
AS AUTO CDC INTO employees
FROM stream(bronze.employee_changes)
KEYS (employee_id)
APPLY AS DELETE WHEN operation = 'DELETE'
SEQUENCE BY sequence_num
COLUMNS * EXCEPT (operation, sequence_num)
STORED AS SCD TYPE 1;

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

dp.create_streaming_table("employees")

dp.create_auto_cdc_flow(
    target="employees",
    source="employees_cdf",
    keys=["employee_id"],
    sequence_by=col("sequence_num"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "sequence_num"],
    stored_as_scd_type=1
)

Volba SCD typu rozhoduje, co se stane s předchozí verzí záznamu:

• SCD Type 1 — update přepíše řádek; zaměstnanec se přestěhuje ze Seattlu do Portlandu a Seattle z tabulky zmizí.
• SCD Type 2 — historie zůstává: záznam se Seattle dostane __END_AT, přibude nový řádek s Portlandem. Přes TRACK HISTORY ON * EXCEPT (...) lze určit, které sloupce novou historickou verzi vůbec vyvolají.

Structured Streaming pracuje s příchozími daty jako s neohraničenou tabulkou, která neustále roste. API je stejné jako u batchů — jen místo spark.read píšete spark.readStream a místo write pak writeStream. O to, co už bylo zpracováno, se stará checkpoint.

Pro Azure Event Hubs je nejjednodušší cesta přes jeho Kafka-kompatibilní endpoint — stejný konektor jako pro Kafku, jen s SASL autentizací přes connection string ze secrets. Existuje i nativní konektor eventhubs (payload pak najdete ve sloupci body místo value).

# cteni z Kafka topicu a zapis do Delta tabulky
df_stream = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker-server:9092")
    .option("subscribe", "sensor-readings")
    .option("startingOffsets", "latest")
    .load())

(df_stream.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoints/sensor-ingestion")
    .toTable("iot.bronze.sensor_readings"))

Kafka konektor vrací standardní schéma (key, value, topic, partition, offset, timestamp) — hodnoty jsou binární, takže je typicky castujete na string a parsujete přes from_json. Frekvenci zpracování řídí trigger: bez triggeru běží stream nepřetržitě, processingTime="10 seconds" zpracovává v intervalech a availableNow=True zpracuje vše dostupné a skončí — ideální pro plánované joby.

Auto Loader je Structured Streaming source specializovaný na soubory: sleduje lokaci v cloud storage a inkrementálně zpracovává nově příchozí soubory. Stav o už zpracovaných souborech si drží sám, takže každý soubor je ingestován exactly-once — bez ručního trackování, bez duplicit při restartu.

Detekce nových souborů běží ve dvou režimech: directory listing (periodický výpis adresáře, funguje bez další konfigurace) a file notification (notifikace z cloudu při příchodu souboru — nižší latence a náklady na API u velkých workloadů; doporučeno s file events na external location v Unity Catalogu).

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", schema_path)
  .option("cloudFiles.inferColumnTypes", "true")
  .load("abfss://container@storage.dfs.core.windows.net/incoming/orders/")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("sales.bronze.orders"))

-- streaming tabulka v deklarativni pipeline: STREAM zapina Auto Loader
CREATE OR REFRESH STREAMING TABLE bronze_orders
AS SELECT * FROM STREAM read_files(
  'abfss://container@storage.dfs.core.windows.net/incoming/orders/',
  format => 'json',
  schemaHints => 'order_id INT, amount DECIMAL(10,2)'
)

Schema inference: Auto Loader umí schéma odvodit sám a uložit do schemaLocation. U JSON a CSV inferuje defaultně vše jako string — automatické typy zapnete přes cloudFiles.inferColumnTypes, konkrétní sloupce vynutíte přes cloudFiles.schemaHints. Když se ve zdroji objeví nové sloupce, rozhoduje cloudFiles.schemaEvolutionMode:

Sloupec _rescued_data je záchranná síť: zachytí data neodpovídající schématu (typové neshody, neočekávané sloupce), takže se nic tiše neztratí.

Předchozí techniky jsou stavební kameny; deklarativní pipeline je rámec, který z nich složí celou pipeline. Deklarujete co chcete (tabulky a jejich zdroje v SQL nebo Pythonu) a framework řeší jak: pořadí spouštění, paralelismus, retry při selhání, checkpointy i inkrementální zpracování. Cílem ingestu je typicky streaming table — Delta tabulka s vestavěnou podporou streamingu a garancí exactly-once.

from pyspark import pipelines as dp

# streaming tabulka definovana dekoratorem; checkpointy resi pipeline
@dp.table
def orders_bronze():
    return (
        spark.readStream.format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load("abfss://container@storage.dfs.core.windows.net/incoming/orders/")
    )

Kromě výchozího flow (vzniká automaticky s tabulkou) můžete definovat explicitní flows — třeba dva regionální zdroje tekoucí do jedné tabulky bez UNION. Nový zdroj pak přidáte novým flow, aniž byste sahali na existující nebo vyvolali full refresh:

CREATE OR REFRESH STREAMING TABLE orders_us;

CREATE FLOW orders_west_flow
AS INSERT INTO orders_us BY NAME
SELECT * FROM STREAM(orders_west);

CREATE FLOW orders_east_flow
AS INSERT INTO orders_us BY NAME
SELECT * FROM STREAM(orders_east);

Pipelines umí číst i z message busů (read_kafka, Event Hubs přes Kafka rozhraní) a pro semi-strukturovaný JSON nabízejí from_json s automatickou inferencí a evolucí schématu (schéma NULL + schemaLocationKey). Monitoring poskytuje event log a DAG závislostí v Lakeflow Pipelines Editoru.

Ve fiktivní firmě Solaris Energy (solární farmy ve Španělsku a Německu, větrné turbíny v Nizozemsku a Dánsku) budujete centrální platformu pro měření výroby energie. Vytvoříte hierarchii katalog → schéma → volume, načtete CSV s odečty přes PySpark DataFrame, doplníte inkrementální load přes COPY INTO, souhrnnou tabulku přes CTAS a nakonec nastavíte Auto Loader s exactly-once sémantikou pro průběžně přibývající soubory.

▶ Otevřít lab na GitHubu (EN) · Learn modul k tématu — s krátkými výukovými videy