Správa kvality dat s očekáváními kanálu

Pomocí očekávání použijte omezení kvality, která ověřují data při jejich toku prostřednictvím kanálů ETL. Očekávání poskytují lepší přehled o metrikách kvality dat a umožňují selhání aktualizací nebo vyřazení záznamů při zjišťování neplatných záznamů.

Pokročilé případy použití a doporučené osvědčené postupy najdete v tématu Doporučení k očekávání a pokročilé vzory.

Co jsou očekávání?

Očekávání jsou volitelná klauzule v materializovaném zobrazení kanálu, tabulce streamování nebo příkazech pro vytváření zobrazení, které u každého záznamu procházejícím dotazem používají kontroly kvality dat. Očekávání používají standardní logické příkazy SQL k určení omezení. Můžete kombinovat více očekávání pro jednu datovou sadu a nastavit očekávání napříč všemi deklaracemi datových sad v kanálu.

Poznámka:

Můžete také definovat očekávání pro streamovací tabulky a materializovaná zobrazení podporovaná samostatnou pipeline vytvořenou v Databricks SQL. Použijte klauzuli CONSTRAINT expectation_name EXPECT (expectation_expr) in CREATE STREAMING TABLE a CREATE MATERIALIZED VIEW.

Následující části představují tři komponenty očekávání a poskytují příklady syntaxe.

Název očekávání

Každé očekávání musí mít název, který se používá jako identifikátor ke sledování a monitorování očekávání. Zvolte název, který oznamuje ověřované metriky. Následující příklad definuje předpoklad valid_customer_age, že věk je mezi 0 a 120 lety.

Důležité

Pro danou datovou sadu musí být jedinečný název očekávání. Můžete znovu použít očekávání přes více datových sad v rámci pipeline. Podívejte se na přenosná a opakovaně použitelná očekávání.

Python

@dp.table
@dp.expect("valid_customer_age", "age BETWEEN 0 AND 120")
def customers():
  return spark.readStream.table("datasets.samples.raw_customers")

SQL

CREATE OR REFRESH STREAMING TABLE customers(
  CONSTRAINT valid_customer_age EXPECT (age BETWEEN 0 AND 120)
) AS SELECT * FROM STREAM(datasets.samples.raw_customers);

Omezení k vyhodnocení

Klauzule constraint je podmíněný příkaz SQL, který musí být vyhodnocen jako true nebo false pro každý záznam. Omezení obsahuje skutečnou logiku toho, co se ověřuje. Pokud záznam nesplní tuto podmínku, očekávání se spustí.

Omezení musí používat platnou syntaxi SQL a nesmí obsahovat následující:

• Vlastní funkce Pythonu
• Volání externích služeb
• Poddotazy odkazující na jiné tabulky

Tady jsou příklady omezení, která je možné přidat do příkazů pro vytváření datových sad:

Python

Syntaxe omezení v Pythonu je:

@dp.expect(<constraint-name>, <constraint-clause>)

Můžete zadat několik omezení:

@dp.expect(<constraint-name>, <constraint-clause>)
@dp.expect(<constraint2-name>, <constraint2-clause>)

Examples:

# Simple constraint
@dp.expect("non_negative_price", "price >= 0")

# SQL functions
@dp.expect("valid_date", "year(transaction_date) >= 2020")

# CASE statements
@dp.expect("valid_order_status", """
   CASE
     WHEN type = 'ORDER' THEN status IN ('PENDING', 'COMPLETED', 'CANCELLED')
     WHEN type = 'REFUND' THEN status IN ('PENDING', 'APPROVED', 'REJECTED')
     ELSE false
   END
""")

# Multiple constraints
@dp.expect("non_negative_price", "price >= 0")
@dp.expect("valid_purchase_date", "date <= current_date()")

# Complex business logic
@dp.expect(
  "valid_subscription_dates",
  """start_date <= end_date
    AND end_date <= current_date()
    AND start_date >= '2020-01-01'"""
)

# Complex boolean logic
@dp.expect("valid_order_state", """
   (status = 'ACTIVE' AND balance > 0)
   OR (status = 'PENDING' AND created_date > current_date() - INTERVAL 7 DAYS)
""")

SQL

Syntaxe omezení v SQL je:

CONSTRAINT <constraint-name> EXPECT ( <constraint-clause> )

Více omezení musí být oddělena čárkou:

CONSTRAINT <constraint-name> EXPECT ( <constraint-clause> ),
CONSTRAINT <constraint2-name> EXPECT ( <constraint2-clause> )

Examples:

-- Simple constraint
CONSTRAINT non_negative_price EXPECT (price >= 0)

-- SQL functions
CONSTRAINT valid_date EXPECT (year(transaction_date) >= 2020)

-- CASE statements
CONSTRAINT valid_order_status EXPECT (
  CASE
    WHEN type = 'ORDER' THEN status IN ('PENDING', 'COMPLETED', 'CANCELLED')
    WHEN type = 'REFUND' THEN status IN ('PENDING', 'APPROVED', 'REJECTED')
    ELSE false
  END
)

-- Multiple constraints
CONSTRAINT non_negative_price EXPECT (price >= 0),
CONSTRAINT valid_purchase_date EXPECT (date <= current_date())

-- Complex business logic
CONSTRAINT valid_subscription_dates EXPECT (
  start_date <= end_date
  AND end_date <= current_date()
  AND start_date >= '2020-01-01'
)

-- Complex boolean logic
CONSTRAINT valid_order_state EXPECT (
  (status = 'ACTIVE' AND balance > 0)
  OR (status = 'PENDING' AND created_date > current_date() - INTERVAL 7 DAYS)
)

Akce pro neplatný záznam

Je nutné zadat akci, která určí, co se stane, když záznam selže při kontrole ověření. Následující tabulka popisuje dostupné akce:

Činnost	Syntaxe SQL	Syntaxe Pythonu	Result
upozornění (výchozí)	EXPECT	dp.expect	Do cíle se zapisují neplatné záznamy.
přetáhnout	EXPECT ... ON VIOLATION DROP ROW	dp.expect_or_drop	Neplatné záznamy se zahodí před zápisem dat do cílového úložiště. Počet vynechaných záznamů se protokoluje spolu s dalšími metrikami datové sady.
selhat	EXPECT ... ON VIOLATION FAIL UPDATE	dp.expect_or_fail	Neplatné záznamy brání úspěšné aktualizaci. Před opětovnou úpravou je vyžadován ruční zásah. Toto očekávání způsobí selhání jednoho toku a nezpůsobí selhání jiných toků ve vašem kanálu.

Můžete také implementovat pokročilou logiku pro karanténu neplatných záznamů bez selhání nebo vyřazení dat. Viz Neplatné záznamy v karanténě.

Metriky sledování očekávání

Sledovací metriky pro warn nebo drop akce lze zobrazit v uživatelském rozhraní kanálu. Vzhledem k tomu, že fail aktualizace selže při zjištění neplatného záznamu, metriky se nezaznamenávají.

Poznámka:

U streamovaných tabulek a materializovaných zobrazení založených na samostatném kanálu vytvořeném v Databricks SQL není karta Kvalita dat v uživatelském rozhraní kanálu dostupná. Dotazem na protokol událostí zobrazte očekávané metriky. Podívejte se na metriky kvality nebo očekávání dotazů na data.

Pokud chcete zobrazit očekávané metriky, proveďte následující kroky:

1. Na bočním panelu pracovního prostoru Azure Databricks klikněte na Úlohy a kanály.
2. Klikněte na název vašeho potrubí.
3. Klikněte na datovou sadu s definovaným očekáváním.
4. Na pravém bočním panelu vyberte kartu Kvalita dat .

Metriky kvality dat můžete zobrazit dotazováním protokolu událostí deklarativních kanálů Sparku Lakeflow. Podívejte se na metriky kvality nebo očekávání dotazů na data.

Zachování neplatných záznamů

Zachování neplatných záznamů je výchozím chováním pro očekávání. Použijte operátor expect, pokud chcete zachovat záznamy, které porušují očekávání, a zároveň shromažďovat metriky o tom, kolik záznamů projde nebo selže při splnění omezení. Záznamy, které porušují očekávání, se přidají do cílové datové sady spolu s platnými záznamy:

Python

@dp.expect("valid timestamp", "timestamp > '2012-01-01'")

SQL

CONSTRAINT valid_timestamp EXPECT (timestamp > '2012-01-01')

Odstranění neplatných záznamů

Pomocí operátoru expect_or_drop můžete zabránit dalšímu zpracování neplatných záznamů. Záznamy, které porušují očekávání, se z cílové datové sady zahodí:

Python

@dp.expect_or_drop("valid_current_page", "current_page_id IS NOT NULL AND current_page_title IS NOT NULL")

SQL

CONSTRAINT valid_current_page EXPECT (current_page_id IS NOT NULL and current_page_title IS NOT NULL) ON VIOLATION DROP ROW

Selhání u neplatných záznamů

Pokud jsou neplatné záznamy nepřijatelné, použijte operátor expect_or_fail k okamžitému zastavení spuštění, když záznam neprojde ověřením. Pokud je operace aktualizace tabulky, systém atomicky vrátí zpět transakci:

Python

@dp.expect_or_fail("valid_count", "count > 0")

SQL

CONSTRAINT valid_count EXPECT (count > 0) ON VIOLATION FAIL UPDATE

Důležité

Pokud máte v kanálu definovaných více paralelních toků, selhání jednoho toku nezpůsobí selhání jiných toků.

Řešení potíží s neúspěšnými aktualizacemi z očekávání

Pokud kanál selže z důvodu porušení očekávání, musíte před opětovným spuštěním kanálu opravit kód kanálu, který zpracuje neplatná data správně.

Očekávání nakonfigurovaná pro neúspěšné pipeline modifikují Spark query plán vašich transformací, aby sledovaly informace potřebné pro odhalování a hlášení porušení. Tyto informace můžete použít k identifikaci toho, který vstupní záznam způsobil porušení mnoha dotazů. Deklarativní kanály Sparku pro Lakeflow poskytují vyhrazenou chybovou zprávu pro hlášení takových porušení. Tady je příklad chybové zprávy o porušení očekávání:

[EXPECTATION_VIOLATION.VERBOSITY_ALL] Flow 'sensor-pipeline' failed to meet the expectation. Violated expectations: 'temperature_in_valid_range'. Input data: '{"id":"TEMP_001","temperature":-500,"timestamp_ms":"1710498600"}'. Output record: '{"sensor_id":"TEMP_001","temperature":-500,"change_time":"2024-03-15 10:30:00"}'. Missing input data: false

Řiďte různá očekávání

Poznámka:

I když SQL i Python podporují v jedné datové sadě více očekávání, umožňuje pouze Python seskupit více očekávání a zadat kolektivní akce.

Můžete seskupit více očekávání dohromady a určit kolektivní akce pomocí funkcí expect_all, expect_all_or_dropa expect_all_or_fail.

Tyto dekorátory přijímají slovník Pythonu jako argument, kde klíč je očekávaný název a hodnota je omezení očekávání. Stejnou sadu očekávání můžete použít v několika datových sadách ve vašem pracovním postupu. Následující příklady ukazují jednotlivé operátory Pythonu expect_all :

valid_pages = {"valid_count": "count > 0", "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"}

@dp.table
@dp.expect_all(valid_pages)
def raw_data():
  # Create a raw dataset

@dp.table
@dp.expect_all_or_drop(valid_pages)
def prepared_data():
  # Create a cleaned and prepared dataset

@dp.table
@dp.expect_all_or_fail(valid_pages)
def customer_facing_data():
  # Create cleaned and prepared to share the dataset

omezení

• Vzhledem k tomu, že očekávání podporují pouze streamované tabulky, materializovaná zobrazení a dočasná zobrazení, podporují se metriky kvality dat pouze pro tyto typy objektů.
• Metriky kvality dat nejsou k dispozici, pokud:
  • V dotazu nejsou definována žádná očekávání.
  • Tok používá operátor, který nepodporuje očekávání.
  • Typ toku, například jímky, nepodporuje očekávání.
  • Pro dané spuštění toku nejsou k dispozici žádné aktualizace přidružené tabulky streamování nebo materializovaného zobrazení.
  • Konfigurace kanálu nezahrnuje potřebná nastavení pro zachytávání metrik, například pipelines.metrics.flowTimeReporter.enabled.
• V některých případech tok COMPLETED nemusí obsahovat metriky. Místo toho se metriky hlásí v každé mikrodávce v události flow_progress se stavem RUNNING.
• Vzhledem k tomu, že zobrazení se počítají pouze při dotazování, nemusí být metriky kvality dat dostupné pro definované zobrazení. Případně může mít zobrazení, které se dotazuje v několika podřízených datových sadách, několik sad metrik kvality dat.
• Očekávání nejsou podporována s AUTO CDC FROM SNAPSHOT.