Share via

Gestire la qualità dei dati con le tabelle live Delta

  • Articolo

Si usano le aspettative per definire i vincoli di qualità dei dati sul contenuto di un set di dati. Le aspettative consentono di garantire che i dati in arrivo nelle tabelle soddisfino i requisiti di qualità dei dati e forniscano informazioni dettagliate sulla qualità dei dati per ogni aggiornamento della pipeline. Si applicano aspettative alle query usando elementi decorator Python o clausole di vincolo SQL.

Quali sono le aspettative delle tabelle live delta?

Le aspettative sono clausole facoltative aggiunte alle dichiarazioni del set di dati Delta Live Tables che applicano controlli di qualità dei dati per ogni record che passano attraverso una query.

Un'aspettativa è costituita da tre cose:

  • Descrizione, che funge da identificatore univoco e consente di tenere traccia delle metriche per il vincolo.
  • Istruzione booleana che restituisce sempre true o false in base a una condizione dichiarata.
  • Azione da eseguire quando un record non soddisfa le aspettative, ovvero il valore booleano restituisce false.

La matrice seguente mostra le tre azioni che è possibile applicare ai record non validi:

Azione Risultato
warn (impostazione predefinita) I record non validi vengono scritti nella destinazione; l'errore viene segnalato come metrica per il set di dati.
Goccia I record non validi vengono eliminati prima che i dati vengano scritti nella destinazione; l'errore viene segnalato come metriche per il set di dati.
fail I record non validi impediscono l'esito positivo dell'aggiornamento. L'intervento manuale è necessario prima della rielaborazione.

È possibile visualizzare le metriche relative alla qualità dei dati, ad esempio il numero di record che violano le aspettative eseguendo una query sul registro eventi delle tabelle live Delta. Vedere Monitorare le pipeline delle tabelle live delta.

Per un riferimento completo alla sintassi della dichiarazione del set di dati Delta Live Tables, vedere Informazioni di riferimento sul linguaggio Python per tabelle live Delta o informazioni di riferimento sul linguaggio SQL per le tabelle live Delta.

Nota

Sebbene sia possibile includere più clausole in qualsiasi previsione, solo Python supporta la definizione di azioni in base a più aspettative. Vedere Multiple expectations (Aspettative multiple).

Conservare record non validi

Usare l'operatore expect quando si desidera mantenere i record che violano le aspettative. I record che violano le aspettative vengono aggiunti al set di dati di destinazione insieme ai record validi:

Python

@dlt.expect("valid timestamp", "col(“timestamp”) > '2012-01-01'")

SQL

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

Eliminare record non validi

Utilizzare l'operatore expect or drop per impedire un'ulteriore elaborazione di record non validi. I record che violano le aspettative vengono eliminati dal set di dati di destinazione:

Python

@dlt.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

Errore nei record non validi

Se i record non validi non sono accettabili, usare l'operatore per arrestare immediatamente l'esecuzione expect or fail quando un record non riesce la convalida. Se l'operazione è un aggiornamento della tabella, il sistema esegue il rollback della transazione in modo atomico:

Python

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

SQL

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

Quando una pipeline ha esito negativo a causa di una violazione delle aspettative, è necessario correggere il codice della pipeline per gestire correttamente i dati non validi prima di eseguire nuovamente la pipeline.

Le aspettative di errore modificano il piano di query Spark delle trasformazioni per tenere traccia delle informazioni necessarie per rilevare e segnalare violazioni. Per molte query, è possibile usare queste informazioni per identificare il record di input che ha generato la violazione. Di seguito è riportata un'eccezione di esempio:

Expectation Violated:
{
  "flowName": "a-b",
  "verboseInfo": {
    "expectationsViolated": [
      "x1 is negative"
    ],
    "inputData": {
      "a": {"x1": 1,"y1": "a },
      "b": {
        "x2": 1,
        "y2": "aa"
      }
    },
    "outputRecord": {
      "x1": 1,
      "y1": "a",
      "x2": 1,
      "y2": "aa"
    },
    "missingInputData": false
  }
}

Aspettative multiple

È possibile definire le aspettative con uno o più vincoli di qualità dei dati nelle pipeline Python. Questi elementi decorator accettano un dizionario Python come argomento, dove la chiave è il nome previsto e il valore è il vincolo di attesa.

Usare expect_all per specificare più vincoli di qualità dei dati quando i record che non superano la convalida devono essere inclusi nel set di dati di destinazione:

@dlt.expect_all({"valid_count": "count > 0", "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"})

Usare expect_all_or_drop per specificare più vincoli di qualità dei dati quando i record che non superano la convalida devono essere eliminati dal set di dati di destinazione:

@dlt.expect_all_or_drop({"valid_count": "count > 0", "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"})

Usare expect_all_or_fail per specificare più vincoli di qualità dei dati quando i record che non superano la convalida devono interrompere l'esecuzione della pipeline:

@dlt.expect_all_or_fail({"valid_count": "count > 0", "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"})

È anche possibile definire una raccolta di aspettative come variabile e passarla a una o più query nella pipeline:

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

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

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

Mettere in quarantena dati non validi

Nell'esempio seguente vengono utilizzate le aspettative in combinazione con tabelle e viste temporanee. Questo modello fornisce metriche per i record che superano i controlli delle aspettative durante gli aggiornamenti della pipeline e consente di elaborare record validi e non validi tramite percorsi downstream diversi.

Nota

Questo esempio legge i dati di esempio inclusi nei set di dati di Databricks. Poiché i set di dati di Databricks non sono supportati con una pipeline che pubblica in Unity Catalog, questo esempio funziona solo con una pipeline configurata per la pubblicazione nel metastore Hive. Tuttavia, questo modello funziona anche con le pipeline abilitate per Unity Catalog, ma è necessario leggere i dati da posizioni esterne. Per altre informazioni sull'uso del catalogo Unity con Delta Live Tables, vedere Usare il catalogo Unity con le pipeline Delta Live Tables.

import dlt
from pyspark.sql.functions import expr

rules = {}
rules["valid_website"] = "(Website IS NOT NULL)"
rules["valid_location"] = "(Location IS NOT NULL)"
quarantine_rules = "NOT({0})".format(" AND ".join(rules.values()))

@dlt.table(
  name="raw_farmers_market"
)
def get_farmers_market_data():
  return (
    spark.read.format('csv').option("header", "true")
      .load('/databricks-datasets/data.gov/farmers_markets_geographic_data/data-001/')
  )

@dlt.table(
  name="farmers_market_quarantine",
  temporary=True,
  partition_cols=["is_quarantined"]
)
@dlt.expect_all(rules)
def farmers_market_quarantine():
  return (
    dlt.read("raw_farmers_market")
      .select("MarketName", "Website", "Location", "State",
              "Facebook", "Twitter", "Youtube", "Organic", "updateTime")
      .withColumn("is_quarantined", expr(quarantine_rules))
  )

@dlt.view(
  name="valid_farmers_market"
)
def get_valid_farmers_market():
  return (
    dlt.read("farmers_market_quarantine")
      .filter("is_quarantined=false")
  )

@dlt.view(
  name="invalid_farmers_market"
)
def get_invalid_farmers_market():
  return (
    dlt.read("farmers_market_quarantine")
      .filter("is_quarantined=true")
  )

Convalidare i conteggi delle righe tra tabelle

È possibile aggiungere una tabella aggiuntiva alla pipeline che definisce un'aspettativa per confrontare i conteggi delle righe tra due tabelle live. I risultati di questa aspettativa vengono visualizzati nel registro eventi e nell'interfaccia utente delle tabelle live delta. In questo esempio vengono convalidati i conteggi di righe uguali tra le tbla tabelle e tblb :

CREATE OR REFRESH LIVE TABLE count_verification(
  CONSTRAINT no_rows_dropped EXPECT (a_count == b_count)
) AS SELECT * FROM
  (SELECT COUNT(*) AS a_count FROM LIVE.tbla),
  (SELECT COUNT(*) AS b_count FROM LIVE.tblb)

Eseguire la convalida avanzata con le aspettative delle tabelle live Delta

È possibile definire tabelle live usando query di aggregazione e join e usare i risultati di tali query come parte del controllo delle aspettative. Ciò è utile se si desidera eseguire controlli di qualità dei dati complessi, ad esempio assicurando che una tabella derivata contenga tutti i record della tabella di origine o garantendo l'uguaglianza di una colonna numerica tra le tabelle. È possibile usare la TEMPORARY parola chiave per impedire la pubblicazione di queste tabelle nello schema di destinazione.

Nell'esempio seguente viene verificato che nella tabella siano presenti report tutti i record previsti:

CREATE TEMPORARY LIVE TABLE report_compare_tests(
  CONSTRAINT no_missing_records EXPECT (r.key IS NOT NULL)
)
AS SELECT * FROM LIVE.validation_copy v
LEFT OUTER JOIN LIVE.report r ON v.key = r.key

Nell'esempio seguente viene usata un'aggregazione per garantire l'univocità di una chiave primaria:

CREATE TEMPORARY LIVE TABLE report_pk_tests(
  CONSTRAINT unique_pk EXPECT (num_entries = 1)
)
AS SELECT pk, count(*) as num_entries
FROM LIVE.report
GROUP BY pk

Rendere le aspettative portabili e riutilizzabili

È possibile gestire le regole di qualità dei dati separatamente dalle implementazioni della pipeline.

Databricks consiglia di archiviare le regole in una tabella Delta con ogni regola categorizzata da un tag. Questo tag viene usato nelle definizioni dei set di dati per determinare le regole da applicare.

Nell'esempio seguente viene creata una tabella denominata rules per gestire le regole:

CREATE OR REPLACE TABLE
  rules
AS SELECT
  col1 AS name,
  col2 AS constraint,
  col3 AS tag
FROM (
  VALUES
  ("website_not_null","Website IS NOT NULL","validity"),
  ("location_not_null","Location IS NOT NULL","validity"),
  ("state_not_null","State IS NOT NULL","validity"),
  ("fresh_data","to_date(updateTime,'M/d/yyyy h:m:s a') > '2010-01-01'","maintained"),
  ("social_media_access","NOT(Facebook IS NULL AND Twitter IS NULL AND Youtube IS NULL)","maintained")
)

L'esempio python seguente definisce le aspettative sulla qualità dei dati in base alle regole archiviate nella rules tabella. La get_rules() funzione legge le regole dalla rules tabella e restituisce un dizionario Python contenente regole corrispondenti all'argomento tag passato alla funzione. Il dizionario viene applicato negli @dlt.expect_all_*() elementi Decorator per applicare vincoli di qualità dei dati. Ad esempio, tutti i record con cui si verificano errori nelle regole contrassegnate validity verranno eliminati dalla raw_farmers_market tabella:

Nota

Questo esempio legge i dati di esempio inclusi nei set di dati di Databricks. Poiché i set di dati di Databricks non sono supportati con una pipeline che pubblica in Unity Catalog, questo esempio funziona solo con una pipeline configurata per la pubblicazione nel metastore Hive. Tuttavia, questo modello funziona anche con le pipeline abilitate per Unity Catalog, ma è necessario leggere i dati da posizioni esterne. Per altre informazioni sull'uso del catalogo Unity con Delta Live Tables, vedere Usare il catalogo Unity con le pipeline Delta Live Tables.

import dlt
from pyspark.sql.functions import expr, col

def get_rules(tag):
  """
    loads data quality rules from a table
    :param tag: tag to match
    :return: dictionary of rules that matched the tag
  """
  rules = {}
  df = spark.read.table("rules")
  for row in df.filter(col("tag") == tag).collect():
    rules[row['name']] = row['constraint']
  return rules

@dlt.table(
  name="raw_farmers_market"
)
@dlt.expect_all_or_drop(get_rules('validity'))
def get_farmers_market_data():
  return (
    spark.read.format('csv').option("header", "true")
      .load('/databricks-datasets/data.gov/farmers_markets_geographic_data/data-001/')
  )

@dlt.table(
  name="organic_farmers_market"
)
@dlt.expect_all_or_drop(get_rules('maintained'))
def get_organic_farmers_market():
  return (
    dlt.read("raw_farmers_market")
      .filter(expr("Organic = 'Y'"))
      .select("MarketName", "Website", "State",
        "Facebook", "Twitter", "Youtube", "Organic",
        "updateTime"
      )
  )

Anziché creare una tabella denominata rules per gestire le regole, è possibile creare un modulo Python per le regole principali, ad esempio in un file denominato rules_module.py nella stessa cartella del notebook:

def get_rules_as_list_of_dict():
  return [
    {
      "name": "website_not_null",
      "constraint": "Website IS NOT NULL",
      "tag": "validity"
    },
    {
      "name": "location_not_null",
      "constraint": "Location IS NOT NULL",
      "tag": "validity"
    },
    {
      "name": "state_not_null",
      "constraint": "State IS NOT NULL",
      "tag": "validity"
    },
    {
      "name": "fresh_data",
      "constraint": "to_date(updateTime,'M/d/yyyy h:m:s a') > '2010-01-01'",
      "tag": "maintained"
    },
    {
      "name": "social_media_access",
      "constraint": "NOT(Facebook IS NULL AND Twitter IS NULL AND Youtube IS NULL)",
      "tag": "maintained"
    }
  ]

Modificare quindi il notebook precedente importando il modulo e modificando la get_rules() funzione per leggere dal modulo anziché dalla rules tabella:

import dlt
from rules_module import *
from pyspark.sql.functions import expr, col

df = spark.createDataFrame(get_rules_as_list_of_dict())

def get_rules(tag):
  """
    loads data quality rules from a table
    :param tag: tag to match
    :return: dictionary of rules that matched the tag
  """
  rules = {}
  for row in df.filter(col("tag") == tag).collect():
    rules[row['name']] = row['constraint']
  return rules

@dlt.table(
  name="raw_farmers_market"
)
@dlt.expect_all_or_drop(get_rules('validity'))
def get_farmers_market_data():
  return (
    spark.read.format('csv').option("header", "true")
      .load('/databricks-datasets/data.gov/farmers_markets_geographic_data/data-001/')
  )

@dlt.table(
  name="organic_farmers_market"
)
@dlt.expect_all_or_drop(get_rules('maintained'))
def get_organic_farmers_market():
  return (
    dlt.read("raw_farmers_market")
      .filter(expr("Organic = 'Y'"))
      .select("MarketName", "Website", "State",
        "Facebook", "Twitter", "Youtube", "Organic",
        "updateTime"
      )
  )