Pular para o conteúdo principal

Recomendações de expectativas e padrões avançados

Este artigo contém recomendações para implementar expectativas em escala e exemplos de padrões avançados suportados por expectativas. Esses padrões usam vários conjuntos de dados em conjunto com expectativas e exigem que os usuários entendam a sintaxe e a semântica da visualização materializada, tabelas de transmissão e expectativas.

Para uma visão geral básica do comportamento e da sintaxe das expectativas, consulte gerenciar a qualidade dos dados com expectativas pipeline.

Expectativas portáteis e reutilizáveis​

A Databricks recomenda as seguintes práticas recomendadas ao implementar expectativas para melhorar a portabilidade e reduzir os encargos de manutenção:

Recomendação

Impacto

Armazene as definições de expectativa separadamente da lógica do pipeline.

Aplique facilmente expectativas a vários conjuntos de dados ou pipelines. Atualize, audite e mantenha expectativas sem modificar o código-fonte do pipeline.

Adicione tags personalizadas para criar grupos de expectativas relacionadas.

Filtre expectativas com base em tags.

Aplique expectativas de forma consistente em conjuntos de dados semelhantes.

Use as mesmas expectativas em vários conjuntos de dados e pipelines para avaliar lógica idêntica.

Os exemplos a seguir demonstram o uso de uma tabela ou dicionário Delta para criar um repositório central de expectativas. As funções Python personalizadas então aplicam essas expectativas ao conjunto de dados em um pipeline de exemplo:

O exemplo seguinte cria uma tabela denominada rules para manter as regras:

SQL
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"),
  ("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")
)

O exemplo Python a seguir define expectativas de qualidade de dados com base nas regras na tabela rules . A função get_rules() lê as regras da tabela rules e retorna um dicionário Python contendo regras correspondentes ao argumento tag passado para a função.

Neste exemplo, o dicionário é aplicado usando decoradores @dp.expect_all_or_drop() para impor restrições de qualidade de dados.

Por exemplo, quaisquer registros que não passarem nas tags de regras com validity serão descartados da tabela raw_farmers_market :

Python
from pyspark import pipelines as dp
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
  """
  df = spark.read.table("rules").filter(col("tag") == tag).collect()
  return {
      row['name']: row['constraint']
      for row in df
  }

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

@dp.table
@dp.expect_all_or_drop(get_rules('maintained'))
def organic_farmers_market():
  return (
    spark.read.table("raw_farmers_market")
      .filter(expr("Organic = 'Y'"))
  )

Validação de contagem de linhas​

O exemplo a seguir valida a igualdade da contagem de linhas entre table_a e table_b para verificar se nenhum dado é perdido durante as transformações:

Gráfico de validação de contagem de linhas do pipeline declarativo LakeFlow com expectativas de uso

Python
@dp.view(
  name="count_verification",
  comment="Validates equal row counts between tables"
)
@dp.expect_or_fail("no_rows_dropped", "a_count == b_count")
def validate_row_counts():
  return spark.sql("""
    SELECT * FROM
      (SELECT COUNT(*) AS a_count FROM table_a),
      (SELECT COUNT(*) AS b_count FROM table_b)""")

Detecção de registro ausente​

O seguinte exemplo valida que todos os registros esperados estão presentes na tabela report:

Gráfico de detecção de linhas ausentes do pipeline declarativo LakeFlow com expectativas de uso

Python
@dp.view(
  name="report_compare_tests",
  comment="Validates no records are missing after joining"
)
@dp.expect_or_fail("no_missing_records", "r_key IS NOT NULL")
def validate_report_completeness():
  return (
    spark.read.table("validation_copy").alias("v")
      .join(
        spark.read.table("report").alias("r"),
        on="key",
        how="left_outer"
      )
      .select(
        "v.*",
        "r.key as r_key"
      )
  )

Unicidade key primária​

O exemplo a seguir valida restrições key primária em todas as tabelas:

Gráfico de exclusividade key primária do pipeline declarativo LakeFlow com expectativas de uso

Python
@dp.view(
  name="report_pk_tests",
  comment="Validates primary key uniqueness"
)
@dp.expect_or_fail("unique_pk", "num_entries = 1")
def validate_pk_uniqueness():
  return (
    spark.read.table("report")
      .groupBy("pk")
      .count()
      .withColumnRenamed("count", "num_entries")
  )

Padrão de evolução do esquema​

O exemplo a seguir mostra como lidar com a evolução do esquema para colunas adicionais. Use este padrão ao migrar fonte de dados ou manipular múltiplas versões de dados upstream, garantindo compatibilidade com versões anteriores e reforçando a qualidade dos dados:

LakeFlow Declarative pipeline evolução do esquema de validação com uso de expectativas

Python
@dp.table
@dp.expect_all_or_fail({
  "required_columns": "col1 IS NOT NULL AND col2 IS NOT NULL",
  "valid_col3": "CASE WHEN col3 IS NOT NULL THEN col3 > 0 ELSE TRUE END"
})
def evolving_table():
  # Legacy data (V1 schema)
  legacy_data = spark.read.table("legacy_source")

  # New data (V2 schema)
  new_data = spark.read.table("new_source")

  # Combine both sources
  return legacy_data.unionByName(new_data, allowMissingColumns=True)

Padrão de validação baseado em intervalo​

O exemplo a seguir demonstra como validar novos pontos de dados em relação a intervalos estatísticos históricos, ajudando a identificar discrepâncias e anomalias no seu fluxo de dados:

Validação baseada em intervalo de pipeline declarativo LakeFlow com uso de expectativas

Python
@dp.view
def stats_validation_view():
  # Calculate statistical bounds from historical data
  bounds = spark.sql("""
    SELECT
      avg(amount) - 3 * stddev(amount) as lower_bound,
      avg(amount) + 3 * stddev(amount) as upper_bound
    FROM historical_stats
    WHERE
      date >= CURRENT_DATE() - INTERVAL 30 DAYS
  """)

  # Join with new data and apply bounds
  return spark.read.table("new_data").crossJoin(bounds)

@dp.table
@dp.expect_or_drop(
  "within_statistical_range",
  "amount BETWEEN lower_bound AND upper_bound"
)
def validated_amounts():
  return spark.read.table("stats_validation_view")

Colocar em quarentena registros inválidos​

Esse padrão combina expectativas com tabelas temporárias e visualizações para rastrear métricas de qualidade de dados durante atualizações pipeline e permitir caminhos de processamento separados para registros válidos e inválidos em operações posteriores.

Padrão de quarentena de dados de pipeline declarativo LakeFlow com uso de expectativas

Python
from pyspark import pipelines as dp
from pyspark.sql.functions import expr

rules = {
  "valid_pickup_zip": "(pickup_zip IS NOT NULL)",
  "valid_dropoff_zip": "(dropoff_zip IS NOT NULL)",
}
quarantine_rules = "NOT({0})".format(" AND ".join(rules.values()))

@dp.view
def raw_trips_data():
  return spark.readStream.table("samples.nyctaxi.trips")

@dp.table(
  temporary=True,
  partition_cols=["is_quarantined"],
)
@dp.expect_all(rules)
def trips_data_quarantine():
  return (
    spark.readStream.table("raw_trips_data").withColumn("is_quarantined", expr(quarantine_rules))
  )

@dp.view
def valid_trips_data():
  return spark.read.table("trips_data_quarantine").filter("is_quarantined=false")

@dp.view
def invalid_trips_data():
  return spark.read.table("trips_data_quarantine").filter("is_quarantined=true")
Última atualização em