Gérer la qualité des données avec Delta Live Tables

Vous utilisez les attentes pour définir les contraintes de qualité des données sur le contenu d’un jeu de données. Les attentes vous permettent de garantir que les données arrivant dans des tables répondent aux exigences de qualité des données et fournissent des insights sur la qualité des données pour chaque mise à jour de pipeline. Vous appliquez des attentes aux requêtes à l’aide d’éléments décoratifs Python ou de clauses de contrainte SQL.

Quelles sont les attentes Delta Live Tables ?

Les attentes sont des clauses facultatives que vous ajoutez aux déclarations de jeu de données Delta Live Tables qui appliquent des vérifications de qualité des données sur chaque enregistrement passant par une requête.

Une attente se compose de trois éléments :

• Une description, qui agit en tant qu’identificateur unique et vous permet de suivre les métriques de la contrainte.
• Une instruction booléenne qui retourne toujours true ou false en fonction d’une condition indiquée.
• Une action à entreprendre lorsqu’un enregistrement ne répond pas à l’attente, ce qui signifie que l’instruction booléenne retourne false.

La matrice suivante montre les trois actions que vous pouvez appliquer aux enregistrements non valides :

Vous pouvez afficher les mesures de qualité des données, telles que le nombre d’enregistrements qui violent une attente, en interrogeant le journal des événements de Delta Live Tables. Consultez Surveiller les pipelines Delta Live Tables.

Pour obtenir une référence complète de la syntaxe de déclaration de jeu de données Delta Live Tables, consultez la référence du langage Python Delta Live Tables ou la référence du langage SQL Delta Live Tables.

Conserver les enregistrements non valides

Utilisez l’opérateur expect lorsque vous souhaitez conserver les enregistrements qui violent l’attente. Les enregistrements qui violent l’attente sont ajoutés au jeu de données cible avec des enregistrements valides :

Python

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

SQL

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

Supprimer les enregistrements non valides

Utilisez l’opérateur expect or drop pour empêcher le traitement ultérieur des enregistrements non valides. Les enregistrements qui violent l’attente sont supprimés du jeu de données cible :

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

Échec sur les enregistrements non valides

Lorsque des enregistrements non valides sont inacceptables, utilisez l’opérateur expect or fail pour arrêter immédiatement l’exécution lorsque la validation de l’enregistrement échoue. Si l’opération est une mise à jour de table, le système restaure atomiquement la transaction :

Python

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

SQL

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

Lorsqu’un pipeline échoue en raison d’une violation des attentes, vous devez corriger le code du pipeline pour gérer correctement les données non valides avant de réexécuter le pipeline.

Les attentes d’échec modifient le plan de requête Spark de vos transformations pour suivre les informations nécessaires à la détection et au signalement des violations. Pour de nombreuses requêtes, vous pouvez utiliser ces informations pour identifier l’enregistrement d’entrée qui a provoqué la violation. Voici un exemple d’exception :

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

Attentes multiples

Vous pouvez définir des attentes avec une ou plusieurs contraintes de qualité des données dans des pipelines Python. Ces éléments décoratifs acceptent un dictionnaire Python comme argument, où la clé est le nom de l’attente et la valeur est la contrainte de l’attente.

Utilisez expect_all pour spécifier plusieurs contraintes de qualité des données lorsque les enregistrements dont la validation échoue doivent être inclus dans le jeu de données cible :

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

Utilisez expect_all_or_drop pour spécifier plusieurs contraintes de qualité des données lorsque les enregistrements dont la validation échoue doivent être supprimés du jeu de données cible :

@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"})

Utilisez expect_all_or_fail pour spécifier plusieurs contraintes de qualité des données lorsque les enregistrements dont la validation échoue doivent interrompre l’exécution du 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"})

Vous pouvez également définir une collection d’attentes en tant que variable et la transmettre à une ou plusieurs requêtes de votre 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

Mettre en quarantaine des données non valides

L’exemple suivant utilise des attentes en combinaison avec des tables et des vues temporaires. Ce modèle vous fournit des métriques pour les enregistrements qui passent des vérifications d’attente pendant les mises à jour du pipeline et fournit un moyen de traiter des enregistrements valides et non valides via différents chemins en aval.

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

Valider le nombre de lignes dans les tables

Vous pouvez ajouter une table supplémentaire à votre pipeline qui définit une attente pour comparer les nombres de lignes entre deux vues matérialisées ou tables de diffusion en continu. Les résultats de cette attente s’affichent dans le journal des événements et l’interface utilisateur des tables dynamiques Delta. L’exemple suivant valide les nombres de lignes égaux entre les tables tbla et tblb :

CREATE OR REFRESH MATERIALIZED VIEW 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)

Effectuer une validation avancée avec des attentes Delta Live Tables

Vous pouvez définir des vues matérialisées à l’aide de requêtes d’agrégation et de jointure et utiliser les résultats de ces requêtes dans le cadre de votre vérification des attentes. Cela est utile si vous souhaitez effectuer des vérifications complexes de la qualité des données, par exemple pour vous assurer qu’une table dérivée contient tous les enregistrements de la table source ou pour garantir l’égalité d’une colonne numérique entre les tables.

L’exemple suivant vérifie que tous les enregistrements attendus sont présents dans la table report :

CREATE MATERIALIZED VIEW 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

L’exemple suivant utilise un agrégat pour garantir l’unicité d’une clé primaire :

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

Rendre les attentes portables et réutilisables

Vous pouvez gérer les règles de qualité des données séparément de vos implémentations de pipeline.

Databricks recommande de stocker les règles dans une table Delta avec chaque règle catégorisée par une balise. Vous utilisez cette balise dans les définitions de jeu de données pour déterminer les règles à appliquer.

L’exemple suivant crée une table nommée rules pour gérer les règles :

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’exemple Python suivant définit des attentes de qualité des données en fonction des règles stockées dans la table rules. La fonction get_rules() lit les règles à partir de la table rules et retourne un dictionnaire Python contenant des règles correspondant à l’argument tag passé à la fonction. Le dictionnaire est appliqué dans les décorateurs @dlt.expect_all_*() pour appliquer les contraintes de qualité des données. Par exemple, tous les enregistrements qui échouent avec les règles marquées avec validity sont supprimés de la table raw_farmers_market :

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

Au lieu de créer une table nommée rules pour gérer des règles, vous pouvez créer un module Python pour les gérer, par exemple dans un fichier nommé rules_module.py dans le même dossier que le 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"
    }
  ]

Modifiez ensuite le notebook précédent en important le module et en modifiant la fonction get_rules() pour lire à partir du module, et non à partir de la table rules :

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