Implementar verificações de validação

Os pipelines de dados geralmente encontram registros com valores ausentes, identificadores duplicados ou valores que estão fora dos intervalos aceitáveis. Sem verificações de validação em vigor, esses problemas de qualidade de dados se propagam downstream, causando análise incorreta, relatórios com falha e decisões comerciais não confiáveis. As verificações de validação identificam esses problemas no momento da ingestão de dados, garantindo que apenas dados de qualidade fluam através do seu pipeline.

Nesta unidade, você aprenderá a implementar verificações de validação quanto à nulidade, à cardinalidade de dados e à verificação de intervalo usando as expectativas de pipelines do Lakeflow e as restrições de tabela Delta Lake.

Entender as abordagens de validação no Azure Databricks

O Azure Databricks fornece dois mecanismos principais para implementar verificações de validação: expectativas de pipeline e restrições de tabela. Cada abordagem atende a cenários diferentes e oferece funcionalidades distintas.

As expectativas de pipeline aplicam validação durante as transformações de dados em pipelines do Lakeflow. As expectativas permitem avisar, eliminar registros invalidados ou interromper o pipeline quando os dados violarem suas regras. Essa abordagem funciona bem para tabelas de streaming e exibições materializadas em que você precisa de controle de qualidade em tempo real.

Restrições de tabela impõem regras diretamente nas tabelas Delta Lake. As restrições rejeitam dados inválidos no momento da gravação, impedindo que registros incorretos entrem em suas tabelas. Essa abordagem atende ao processamento em lote e a cenários que exigem garantias rigorosas de integridade de dados.

Com essas abordagens, você pode validar três dimensões principais da qualidade dos dados: nulidade (garantindo que os valores necessários existam), cardinalidade de dados (verificando a exclusividade quando esperado) e verificação de intervalo (confirmando que os valores estão dentro dos limites aceitáveis).

Implementar verificações de nulidade

A validação de nulabilidade garante que as colunas obrigatórias contenham valores. Considere uma tabela de clientes em que email e customer_id devem sempre ter valores. Você pode implementar essas verificações usando as expectativas no seu pipeline.

Usando o Python, adicione decoradores de expectativa à definição da tabela de streaming:

from pyspark import pipelines as dp

@dp.table
@dp.expect_or_drop("valid_email", "email IS NOT NULL")
@dp.expect_or_drop("valid_customer_id", "customer_id IS NOT NULL")
def customers():
    return spark.readStream.table("raw.customers")

A mesma validação no SQL usa a CONSTRAINT cláusula:

CREATE OR REFRESH STREAMING TABLE customers(
    CONSTRAINT valid_email EXPECT (email IS NOT NULL) ON VIOLATION DROP ROW,
    CONSTRAINT valid_customer_id EXPECT (customer_id IS NOT NULL) ON VIOLATION DROP ROW
) AS SELECT * FROM STREAM(raw.customers);

Para tabelas Delta Lake fora de pipelines, use restrições NOT NULL na definição de tabela:

CREATE TABLE customers (
    customer_id INT NOT NULL,
    email STRING NOT NULL,
    first_name STRING,
    last_name STRING
);

Você também pode adicionar restrições de nulidade a tabelas existentes:

ALTER TABLE customers ALTER COLUMN email SET NOT NULL;

Validar a cardinalidade de dados

A validação de cardinalidade garante que as colunas que devem conter valores exclusivos realmente contenham. Essa verificação é essencial para chaves primárias, identificadores de transação e outros campos em que duplicatas indicam problemas de qualidade de dados.

As expectativas de pipeline podem validar a cardinalidade verificando se há condições que indicam problemas de exclusividade. Por exemplo, você pode verificar se um Número de Previdência Social aparece apenas uma vez por pessoa:

from pyspark.sql.window import Window
from pyspark.sql.functions import count

@dp.table
@dp.expect("unique_ssn_per_person", "ssn_count = 1")
def employees():
    df = spark.table("raw.employees")
    w = Window.partitionBy("ssn")
    return df.withColumn("ssn_count", count("*").over(w))

Para verificações de cardinalidade mais abrangentes, combine as expectativas com a lógica de agregação em sua transformação:

CREATE OR REFRESH MATERIALIZED VIEW order_validation AS
SELECT 
    order_id,
    COUNT(*) as occurrence_count
FROM orders
GROUP BY order_id
HAVING COUNT(*) > 1;

Essa visualização materializada identifica quaisquer IDs de pedido duplicadas, permitindo que você investigue e resolva problemas de cardinalidade.

O Delta Lake dá suporte a restrições de chave primárias que documentam a exclusividade esperada, embora essas restrições sejam informativas e não impostas:

CREATE TABLE orders (
    order_id INT NOT NULL,
    customer_id INT,
    order_date DATE,
    CONSTRAINT orders_pk PRIMARY KEY (order_id)
);

Aplicar verificação de intervalo

A validação de intervalo confirma que valores numéricos, de data e outros estão dentro dos limites aceitáveis. Essa verificação captura erros de entrada de dados, falhas no sistema e problemas de integração que produzem valores fora do intervalo.

Defina as expectativas de intervalo usando operadores de comparação ou a BETWEEN cláusula:

@dp.table
@dp.expect_or_fail("valid_age", "age BETWEEN 0 AND 150")
@dp.expect_or_fail("valid_salary", "salary >= 0")
@dp.expect_or_fail("valid_hire_date", "hire_date <= current_date()")
def employees():
    return spark.readStream.table("raw.employees")

No SQL, aplique as mesmas verificações de intervalo:

CREATE OR REFRESH STREAMING TABLE transactions(
    CONSTRAINT valid_amount EXPECT (amount > 0) ON VIOLATION DROP ROW,
    CONSTRAINT valid_date EXPECT (transaction_date <= current_date()) ON VIOLATION DROP ROW,
    CONSTRAINT valid_quantity EXPECT (quantity BETWEEN 1 AND 10000) ON VIOLATION DROP ROW
) AS SELECT * FROM STREAM(raw.transactions);

Para tabelas Delta Lake, use restrições CHECK para impor intervalos no momento da gravação:

ALTER TABLE employees ADD CONSTRAINT valid_age CHECK (age >= 18 AND age <= 120);
ALTER TABLE transactions ADD CONSTRAINT positive_amount CHECK (amount > 0);

As verificações de intervalo também podem validar regras de negócios que combinam várias condições:

@dp.expect("valid_discount", """
    discount_percent >= 0 
    AND discount_percent <= 100
    AND (discount_percent <= 50 OR customer_tier = 'PREMIUM')
""")

Essa expectativa garante que os descontos permaneçam dentro dos limites e impõe que apenas os clientes premium recebam descontos acima de 50%.

Escolher ações para falhas de validação

Quando a validação falha, você escolhe como o pipeline deve responder. Cada ação atende a diferentes requisitos de negócios e níveis de criticidade de dados.

Use a ação de aviso quando precisar de visibilidade sobre problemas de qualidade de dados sem bloquear o fluxo de dados:

@dp.expect("has_phone", "phone_number IS NOT NULL")

Use drop quando registros inválidos devem ser filtrados silenciosamente:

@dp.expect_or_drop("complete_address", "street IS NOT NULL AND city IS NOT NULL")

Use falha quando a integridade dos dados for crítica e os registros inválidos forem inaceitáveis.

@dp.expect_or_fail("valid_account_balance", "balance >= 0")

Você pode exibir as métricas de expectativa na interface do usuário do pipeline selecionando um conjunto de dados com expectativas e abrindo a guia Qualidade de dados . Essas métricas ajudam você a monitorar as taxas de aprovação de validação e a identificar problemas sistemáticos de qualidade de dados.

Agora que você entende como implementar verificações de validação, você pode garantir que seus pipelines de dados mantenham padrões de qualidade para nulidade, cardinalidade e intervalos de valor.