Como aprendi Apache Spark — Parte 14: Data Quality, Data Contracts e Governança

Nota histórica: esta série revisita minha trajetória em Engenharia de Dados iniciada em 2012 e conecta os fundamentos daquela experiência às práticas e tecnologias modernas. Data Contracts, observabilidade e governança evoluíram significativamente desde então.

Nos primeiros capítulos desta série, concentramos nossa atenção em como o Apache Spark executa o processamento distribuído.

Estudamos Driver, Executors, Jobs, Stages, Tasks, partições, shuffle, memória e tolerância a falhas.

Depois avançamos para dimensionamento, formatos de armazenamento e arquiteturas Lakehouse.

Mas existe uma questão que considero tão importante quanto a performance:

O que acontece quando um pipeline executa perfeitamente, mas entrega dados errados?

Imagine um processamento que termina em cinco minutos, utiliza poucos recursos, não apresenta falhas e grava todos os arquivos esperados.

Tecnicamente, a execução foi bem-sucedida.

Entretanto, o dataset final contém pedidos duplicados, valores negativos indevidos, datas inconsistentes e registros sem identificadores.

O Spark executou corretamente as instruções recebidas.

Mas o produto de dados está incorreto.

Essa diferença nos leva a três disciplinas fundamentais:

• Data Quality: verificar se os dados atendem aos requisitos definidos.
• Data Contracts: formalizar expectativas e responsabilidades entre produtores e consumidores.
• Data Governance: estabelecer políticas, responsabilidades, controles e rastreabilidade sobre os dados.

Vamos entender como essas disciplinas se conectam e como implementá-las em pipelines Spark.

---

1. Pipeline bem-sucedido não significa dado confiável

Considere um pipeline simples:

Operational Database
        |
        v
    Ingestion
        |
        v
      Spark
        |
        v
     Parquet
        |
        v
   BI Dashboard

O processamento termina sem erros.

Mas a origem enviou registros duplicados.

O Spark pode gravá-los normalmente se não existir nenhuma regra para identificá-los.

Agora imagine que um dashboard utilize esses registros para calcular faturamento.

O resultado estará incorreto.

Esse é um problema de qualidade de dados, não necessariamente de execução distribuída.

Precisamos distinguir:

Technical Success

A aplicação executou conforme as instruções.

Data Quality Success

Os dados resultantes satisfazem as regras de qualidade estabelecidas.

Business Success

Os dados permitem decisões corretas e atendem às necessidades dos consumidores.

Uma plataforma madura precisa avaliar essas três dimensões.

---

2. As principais dimensões de Data Quality

Existem diferentes modelos de classificação de qualidade de dados. Para aplicações Spark, considero especialmente úteis estas dimensões:

Dimensão	Pergunta
Completude	Os campos obrigatórios estão preenchidos?
Unicidade	Existem duplicidades indevidas?
Validade	Os valores respeitam os domínios e formatos definidos?
Consistência	Os dados respeitam regras entre campos e sistemas?
Atualidade	Os dados estão disponíveis dentro da janela esperada?
Acurácia	Os valores representam corretamente a realidade?
Integridade referencial	As referências apontam para entidades válidas?

Essas dimensões exigem mecanismos diferentes.

Por exemplo, podemos verificar se um campo está preenchido com Spark.

Mas determinar se um endereço realmente existe pode exigir uma fonte de referência externa.

Portanto, qualidade não é apenas executar expressões isNotNull().

É definir critérios verificáveis para cada contexto.

---

3. Construindo um laboratório de qualidade com PySpark

Vamos utilizar um dataset sintético de pedidos.

O exemplo foi projetado para funcionar em uma instalação PySpark compatível, sem exigir Iceberg, Delta Lake ou bibliotecas externas.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType,
    StructField,
    LongType,
    StringType,
    DecimalType
)
from decimal import Decimal

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("SparkDataQuality")
    .getOrCreate()
)

schema = StructType([
    StructField("pedido_id", LongType(), True),
    StructField("cliente_id", LongType(), True),
    StructField("status", StringType(), True),
    StructField("valor", DecimalType(12, 2), True),
    StructField("data_pedido", StringType(), True)
])

registros = [
    (1, 101, "PAGO", Decimal("150.00"), "2026-11-01"),
    (2, 102, "CRIADO", Decimal("230.00"), "2026-11-02"),
    (2, 102, "CRIADO", Decimal("230.00"), "2026-11-02"),
    (3, None, "PAGO", Decimal("99.90"), "2026-11-03"),
    (4, 104, "DESCONHECIDO", Decimal("50.00"), "2026-11-04"),
    (5, 105, "PAGO", Decimal("-100.00"), "2026-11-05"),
    (6, 106, "PAGO", Decimal("80.00"), "data-invalida")
]

df = spark.createDataFrame(
    registros,
    schema=schema
)

df.show(truncate=False)

Esse dataset contém problemas intencionais:

• Pedido duplicado.
• Cliente ausente.
• Status não permitido.
• Valor negativo.
• Data inválida.

Vamos construir regras para identificá-los.

---

4. Definindo regras de validação

Primeiro, precisamos estabelecer o contrato lógico dos pedidos.

Para este laboratório, vamos considerar:

1. pedido_id obrigatório.
2. cliente_id obrigatório.
3. status pertencente a um domínio conhecido.
4. valor maior ou igual a zero.
5. data_pedido representando uma data válida.
6. pedido_id único dentro do dataset.

Essas são regras de exemplo, não universais.

Em um sistema real, valores negativos podem ser legítimos para estornos ou ajustes.

A regra deve refletir a semântica do domínio.

Vamos implementar:

from pyspark.sql import Window

status_validos = [
    "CRIADO",
    "PAGO",
    "CANCELADO"
]

janela_pedido = Window.partitionBy(
    "pedido_id"
)

df_validado = (
    df
    .withColumn(
        "data_convertida",
        F.expr(
            "try_cast(data_pedido AS DATE)"
        )
    )
    .withColumn(
        "qtd_por_pedido",
        F.count(F.lit(1)).over(janela_pedido)
    )
    .withColumn(
        "erro_pedido_id",
        F.col("pedido_id").isNull()
    )
    .withColumn(
        "erro_cliente_id",
        F.col("cliente_id").isNull()
    )
    .withColumn(
        "erro_status",
        F.col("status").isNull()
        | (~F.col("status").isin(status_validos))
    )
    .withColumn(
        "erro_valor",
        F.col("valor").isNull()
        | (F.col("valor") < F.lit(0))
    )
    .withColumn(
        "erro_data",
        F.col("data_convertida").isNull()
    )
    .withColumn(
        "erro_duplicidade",
        F.col("qtd_por_pedido") > 1
    )
)

Utilizamos try_cast para tratar datas inválidas como nulas sem depender de exceções de conversão. O suporte deve ser confirmado na versão de Spark SQL utilizada.

Para o laboratório, assumimos Spark 3.5 ou posterior.

---

5. Criando uma classificação de qualidade

Agora queremos saber se cada registro é válido.

colunas_erro = [
    "erro_pedido_id",
    "erro_cliente_id",
    "erro_status",
    "erro_valor",
    "erro_data",
    "erro_duplicidade"
]

condicao_invalida = F.lit(False)

for coluna in colunas_erro:
    condicao_invalida = (
        condicao_invalida | F.col(coluna)
    )

df_classificado = (
    df_validado
    .withColumn(
        "registro_valido",
        ~condicao_invalida
    )
    .withColumn(
        "quality_status",
        F.when(
            F.col("registro_valido"),
            F.lit("VALID")
        ).otherwise(
            F.lit("INVALID")
        )
    )
)

df_classificado.select(
    "pedido_id",
    "cliente_id",
    "status",
    "valor",
    "quality_status"
).show()

Agora cada registro possui uma classificação explícita.

Mas precisamos tomar cuidado com a interpretação.

Um registro pode falhar em mais de uma regra.

Portanto, o número de violações não será necessariamente igual ao número de registros inválidos.

Essa distinção é importante para métricas de qualidade.

---

6. Gerando métricas de Data Quality

Vamos calcular indicadores agregados.

metricas = (
    df_classificado
    .agg(
        F.count("*").alias("total_registros"),
        F.sum(
            F.when(
                F.col("registro_valido"),
                1
            ).otherwise(0)
        ).alias("registros_validos"),
        F.sum(
            F.when(
                ~F.col("registro_valido"),
                1
            ).otherwise(0)
        ).alias("registros_invalidos")
    )
    .withColumn(
        "quality_score",
        F.when(
            F.col("total_registros") > 0,
            F.col("registros_validos")
            / F.col("total_registros")
        )
    )
)

metricas.show()

O resultado esperado para o dataset de exemplo é:

Total de registros: 7
Registros válidos: 2
Registros inválidos: 5
Quality Score: aproximadamente 0,2857

A duplicidade torna as duas ocorrências do pedido 2 inválidas.

Esse score é uma métrica operacional simplificada: a proporção de registros que passaram por todas as regras.

Não representa uma medida universal de qualidade.

Em produção, podemos precisar de indicadores separados por dimensão, criticidade e impacto de negócio.

---

7. Data Quality por regra

Um score agregado pode esconder problemas importantes.

Imagine que uma tabela tenha 99% de registros válidos.

Isso parece excelente.

Mas e se o 1% inválido representar justamente os pedidos de maior valor?

Por isso, devemos medir cada regra separadamente.

expressoes = [
    F.count("*").alias("total_registros")
]

for coluna in colunas_erro:
    expressoes.append(
        F.sum(
            F.when(F.col(coluna), 1).otherwise(0)
        ).alias(f"violacoes_{coluna}")
    )

metricas_regras = df_classificado.agg(
    *expressoes
)

metricas_regras.show(
    truncate=False
)

Agora conseguimos identificar quais regras concentram problemas.

Podemos acompanhar tendências como:

Null Customer IDs
Duplicate Orders
Invalid Status
Negative Values
Invalid Dates

Essas métricas são muito mais úteis para diagnóstico do que apenas uma classificação geral.

---

8. Quarentena de registros

O que devemos fazer com registros inválidos?

Uma possibilidade é interromper todo o pipeline.

Outra é separar os registros inválidos para investigação.

Essa segunda estratégia é frequentemente chamada de quarentena.

Conceitualmente:

                 SOURCE DATA
                      |
                      v
                 SPARK VALIDATION
                      |
             +--------+--------+
             |                 |
             v                 v
         VALID DATA        INVALID DATA
             |                 |
             v                 v
       CURATED LAYER      QUARANTINE
             |                 |
             v                 v
         ANALYTICS         INVESTIGATION

Em PySpark:

df_validos = df_classificado.filter(
    F.col("registro_valido")
)

df_quarentena = df_classificado.filter(
    ~F.col("registro_valido")
)

Podemos persistir os resultados:

(
    df_validos.write
    .mode("overwrite")
    .parquet("/tmp/dq/pedidos_validos")
)

(
    df_quarentena.write
    .mode("overwrite")
    .parquet("/tmp/dq/pedidos_quarentena")
)

Esses caminhos são apenas para o laboratório.

Em produção, a quarentena deve ter políticas de acesso, retenção, rastreabilidade e reprocessamento.

---

9. Quarentena não é descarte

Uma decisão importante é preservar informações suficientes para investigar o problema.

Idealmente, um registro em quarentena deve permitir responder:

• Qual fonte produziu o dado?
• Quando ele foi recebido?
• Qual pipeline o processou?
• Qual regra falhou?
• Qual era o valor original?
• O registro pode ser corrigido?
• Como será reprocessado?

Podemos adicionar metadados:

df_quarentena = (
    df_quarentena
    .withColumn(
        "quarantine_timestamp",
        F.current_timestamp()
    )
    .withColumn(
        "pipeline_name",
        F.lit("orders_ingestion")
    )
    .withColumn(
        "validation_version",
        F.lit("1.0.0")
    )
)

Em ambientes corporativos, também devemos registrar um identificador de execução e, quando aplicável, a versão do contrato.

Isso ajuda a conectar qualidade, observabilidade e auditoria.

---

10. Quando interromper o pipeline?

Nem toda violação deve produzir o mesmo comportamento.

Imagine três cenários.

Cenário A — Campo opcional ausente

Podemos aceitar o registro e registrar uma advertência.

Cenário B — Pedido sem identificador

Talvez o registro deva ser enviado para quarentena.

Cenário C — Volume de dados muito abaixo do esperado

Talvez seja necessário interromper a publicação do dataset inteiro.

Podemos definir políticas como:

WARN
  |
  v
Log and Continue

QUARANTINE
  |
  v
Isolate Invalid Records

FAIL
  |
  v
Stop Pipeline / Block Publication

Essa classificação deve fazer parte do desenho do produto de dados.

Não deve depender de decisões improvisadas durante incidentes.

---

11. Quality Gates

Um Quality Gate é uma condição que precisa ser satisfeita antes que os dados avancem para determinada etapa.

Exemplo:

A publicação será bloqueada se menos de 99% dos registros atenderem às regras críticas.

Podemos implementar:

resultado_metricas = metricas.first()

quality_score = resultado_metricas[
    "quality_score"
]

limite_minimo = 0.99

if (
    quality_score is None
    or quality_score < limite_minimo
):
    raise RuntimeError(
        "Quality Gate reprovado: "
        f"score={quality_score}"
    )

No nosso laboratório, a execução será interrompida porque o score está abaixo de 99%.

Isso é esperado.

Entretanto, em produção, um Quality Gate não deve ser a única proteção.

Precisamos garantir que os dados não sejam publicados parcialmente antes da validação.

Uma estratégia é validar em uma área de staging e promover o resultado somente após aprovação.

---

12. O que são Data Contracts?

Até aqui, definimos regras dentro do pipeline.

Mas existe um problema.

Como o sistema produtor sabe quais regras precisa respeitar?

E como os consumidores sabem o que esperar?

É aqui que entram os Data Contracts.

Um Data Contract formaliza expectativas sobre um conjunto de dados e suas interfaces.

Pode incluir:

• Identificação do dataset.
• Responsável pelo dado.
• Schema.
• Tipos.
• Campos obrigatórios.
• Regras de qualidade.
• Semântica.
• Frequência de atualização.
• SLA.
• Política de evolução.
• Classificação de dados.
• Responsabilidades.

Um contrato transforma expectativas implícitas em requisitos explícitos e verificáveis.

---

13. Exemplo de Data Contract

Considere um contrato conceitual em YAML:

contract:
  name: orders
  version: "1.0.0"
  owner: sales-data-team

dataset:
  name: curated.orders
  description: Pedidos comerciais consolidados

schema:
  fields:
    - name: pedido_id
      type: long
      required: true
      unique: true

    - name: cliente_id
      type: long
      required: true

    - name: status
      type: string
      required: true
      allowed_values:
        - CRIADO
        - PAGO
        - CANCELADO

    - name: valor
      type: decimal(12,2)
      required: true
      minimum: 0

    - name: data_pedido
      type: date
      required: true

service_level:
  freshness_minutes: 60

quality:
  minimum_valid_ratio: 0.99
  on_failure: block_publication

Esse YAML é um exemplo didático, não uma implementação validada de um padrão específico de Data Contracts.

Para produção, devemos escolher uma especificação, definir sua versão e implementar validadores compatíveis.

O ponto principal é que as regras deixem de existir apenas como conhecimento informal da equipe.

---

14. Schema Enforcement vs. Data Contract

Esses conceitos estão relacionados, mas não são equivalentes.

Schema Enforcement verifica se os dados respeitam determinadas regras estruturais.

Por exemplo:

pedido_id → LONG
valor     → DECIMAL(12,2)
status    → STRING

Data Contract pode incluir regras adicionais:

pedido_id → obrigatório e único
valor     → não negativo
status    → domínio permitido
freshness → até 60 minutos
owner     → equipe comercial

Portanto:

DATA CONTRACT
      |
      +-- Schema
      +-- Semantics
      +-- Quality Rules
      +-- SLA / SLO
      +-- Ownership
      +-- Evolution Policy

Um schema correto é necessário em muitos cenários, mas não é suficiente para garantir a qualidade de um produto de dados.

---

15. Versionamento e evolução de contratos

Imagine que o produtor adiciona uma coluna:

customer_segment

Talvez essa alteração seja compatível com consumidores existentes.

Agora imagine que o produtor renomeia:

pedido_id

para:

order_identifier

Essa mudança pode quebrar pipelines, dashboards e modelos downstream.

Precisamos de uma política de evolução.

Uma abordagem é classificar alterações em:

• Compatíveis.
• Potencialmente incompatíveis.
• Incompatíveis.

Também podemos utilizar versionamento semântico como convenção de comunicação, desde que as regras sejam explicitamente definidas.

O importante é:

Alterações de contrato devem ser avaliadas pelo impacto nos consumidores, não apenas pela facilidade de implementação no produtor.

---

16. Contract Testing

Um Data Contract só produz valor operacional quando suas regras são verificadas.

Podemos criar testes automatizados para validar:

• Schema.
• Campos obrigatórios.
• Domínios.
• Unicidade.
• Volume.
• Atualidade.
• Compatibilidade entre versões.

Em um pipeline de CI/CD, podemos executar:

Pull Request
     |
     v
Schema Validation
     |
     v
Contract Tests
     |
     v
Data Quality Tests
     |
     v
Integration Tests
     |
     v
Deployment

Isso reduz a probabilidade de alterações incompatíveis chegarem à produção.

Mas testes de CI/CD não substituem validações em runtime.

Os dados reais podem apresentar problemas mesmo quando o código foi aprovado.

---

17. Testes unitários de transformações Spark

Podemos testar uma transformação sem executar um pipeline inteiro.

Exemplo:

from pyspark.sql import functions as F

def filtrar_valores_validos(df):
    return df.filter(
        F.col("valor").isNotNull()
        & (F.col("valor") >= 0)
    )

Um teste simples:

entrada = spark.createDataFrame(
    [
        (1, 100.0),
        (2, -50.0),
        (3, None)
    ],
    ["id", "valor"]
)

resultado = filtrar_valores_validos(
    entrada
)

ids = {
    row["id"]
    for row in resultado.select("id").collect()
}

assert ids == {1}

Esse teste verifica uma regra específica.

Em produção, devemos incluir casos como:

• Valores nulos.
• Tipos inesperados.
• Duplicidades.
• Limites numéricos.
• Datas inválidas.
• Strings vazias.
• Evolução de schema.

---

18. Data Observability

Data Quality verifica propriedades dos dados.

Data Observability procura oferecer visibilidade contínua sobre o comportamento dos dados e pipelines.

Isso pode incluir:

DATA PIPELINE
      |
      +-- Freshness
      +-- Volume
      +-- Schema Changes
      +-- Null Rates
      +-- Distribution Changes
      +-- Pipeline Failures
      +-- Quality Violations

Imagine um pipeline que normalmente recebe 10 milhões de registros por dia.

Hoje recebeu apenas 100 mil.

Os registros podem passar por todas as validações individuais.

Mesmo assim, existe um problema operacional importante.

Precisamos monitorar o comportamento agregado do dataset.

---

19. Freshness e SLA

Imagine que um dashboard comercial precisa ser atualizado a cada hora.

O pipeline termina com sucesso, mas os dados mais recentes possuem três dias.

Tecnicamente, a execução pode ter funcionado.

Operacionalmente, o serviço falhou.

Podemos medir:

[

\text{Data Age}

\text{Current Time}

\text{Latest Event Time}
]

Por exemplo, se a última atualização válida ocorreu há 180 minutos e o limite é 60 minutos, temos uma violação.

É importante distinguir:

• Horário do evento.
• Horário de ingestão.
• Horário de processamento.
• Horário de publicação.

Cada timestamp responde a uma pergunta diferente.

Um pipeline pode processar hoje eventos históricos legítimos.

Por isso, a métrica de freshness deve ser definida conforme a semântica do produto de dados.

---

20. Monitoramento de volume

Podemos acompanhar a quantidade de registros processados por execução.

volume_atual = df.count()

Mas devemos lembrar que count() é uma Action e pode exigir uma varredura distribuída.

Em datasets grandes, precisamos avaliar o custo dessa medição.

Em determinados cenários, métricas da própria execução ou metadados confiáveis podem ser alternativas mais eficientes.

O objetivo é identificar anomalias como:

Expected: 10 million rows
Observed: 100 thousand rows

Uma variação pode indicar:

• Falha de ingestão.
• Filtro incorreto.
• Alteração na origem.
• Partição ausente.
• Problema de conectividade.
• Mudança legítima de comportamento.

O alerta deve iniciar uma investigação, não assumir automaticamente a causa.

---

21. Data Lineage

Agora chegamos a outro componente importante da governança.

Data Lineage representa a origem e as transformações pelas quais os dados passaram.

Imagine:

CRM Database
     |
     v
Raw Customers
     |
     v
Spark Cleaning
     |
     v
Curated Customers
     |
     v
Customer Analytics
     |
     v
Executive Dashboard

Se a coluna customer_segment estiver incorreta, queremos saber:

• Qual sistema produziu o valor?
• Qual transformação o modificou?
• Quais tabelas dependem dele?
• Quais dashboards podem ter sido afetados?
• Quais pipelines precisam ser reexecutados?

Lineage ajuda a responder essas perguntas.

---

22. Lineage não é o mesmo que RDD Lineage

Na Parte 8, estudamos o lineage interno do Spark.

Ele representa dependências computacionais utilizadas, entre outras finalidades, na recuperação de partições.

Agora estamos falando de lineage de dados em nível de plataforma.

A diferença é importante.

RDD Lineage

RDD A
  |
  v
RDD B
  |
  v
RDD C

Data Lineage corporativo

Source System
     |
     v
Ingestion Job
     |
     v
Lakehouse Table
     |
     v
Business Metric
     |
     v
Dashboard

O primeiro está relacionado à execução distribuída.

O segundo está relacionado à rastreabilidade, impacto e governança.

Ambos são importantes, mas resolvem problemas diferentes.

---

23. OpenLineage e integração com pipelines

Uma abordagem para padronizar metadados de lineage é utilizar o projeto OpenLineage.

Ele define um modelo para representar eventos de execução, Jobs e datasets.

Conceitualmente:

Spark Job
    |
    v
Lineage Events
    |
    v
Metadata Collector
    |
    v
Lineage Platform
    |
    v
Impact Analysis

A integração depende das versões e componentes utilizados.

Não devemos assumir que uma instalação padrão do PySpark já produz automaticamente todos os eventos de lineage corporativo.

É necessário configurar mecanismos de coleta e integração compatíveis.

---

24. Catálogo de dados e metadados

Uma plataforma corporativa precisa permitir que usuários descubram e compreendam os datasets disponíveis.

Um catálogo pode registrar informações como:

• Nome.
• Descrição.
• Responsável.
• Schema.
• Classificação.
• Localização.
• Frequência de atualização.
• Regras de qualidade.
• Lineage.
• Termos de negócio.
• Políticas de acesso.

Uma visão conceitual:

                DATA CATALOG
                     |
        +------------+------------+
        |            |            |
        v            v            v
      Schema       Lineage      Ownership
        |            |            |
        +------------+------------+
                     |
                     v
               DATA DISCOVERY

Ferramentas como OpenMetadata e DataHub são exemplos de plataformas que podem apoiar esse tipo de capacidade.

Mas o catálogo não substitui a definição de responsabilidades e processos de governança.

---

25. Governança não é apenas catalogação

É comum reduzir governança à instalação de uma ferramenta.

Mas governança envolve decisões organizacionais.

Precisamos definir:

Data Owner

Responsável pela decisão e pelo uso apropriado dos dados em determinado domínio.

Data Steward

Responsável por apoiar a gestão de definições, qualidade e políticas.

Data Engineering

Responsável pela implementação e operação dos pipelines e estruturas técnicas.

Data Consumer

Responsável por utilizar os dados de acordo com sua finalidade e restrições.

Os papéis exatos variam entre organizações.

O importante é evitar datasets sem responsáveis claros.

---

26. Segurança, privacidade e LGPD

Qualidade e governança também precisam considerar segurança e privacidade.

Imagine um dataset contendo:

customer_id
customer_name
email
phone
document_number
purchase_history

Nem todos os consumidores precisam acessar todas essas informações.

Precisamos avaliar:

• Classificação dos dados.
• Finalidade de tratamento.
• Base legal aplicável.
• Controle de acesso.
• Retenção.
• Minimização.
• Mascaramento ou pseudonimização.
• Auditoria.
• Exclusão ou anonimização quando cabível.

A LGPD não é implementada simplesmente ativando uma configuração do Spark.

Ela exige medidas técnicas, organizacionais e jurídicas adequadas ao contexto.

O Spark participa da arquitetura, mas não substitui uma política de proteção de dados.

---

27. Qualidade em arquiteturas Lakehouse

Na Parte 13, estudamos formatos de tabela como Iceberg, Delta Lake e Hudi.

Agora podemos conectar esses recursos à qualidade de dados.

Uma arquitetura conceitual:

                   DATA SOURCES
                        |
                        v
                    RAW LAYER
                        |
                        v
                 CONTRACT CHECK
                        |
                        v
                DATA QUALITY RULES
                        |
             +----------+----------+
             |                     |
             v                     v
         APPROVED DATA         QUARANTINE
             |                     |
             v                     v
        LAKEHOUSE TABLE        INVESTIGATION
             |
             v
        CONSUMPTION LAYER

O formato de tabela pode ajudar na publicação transacional.

Mas precisamos garantir que as validações sejam realizadas antes de disponibilizar uma versão para os consumidores.

A estratégia exata depende do formato, da engine e do mecanismo de publicação.

---

28. Idempotência e reprocessamento

Imagine que um pipeline falhou depois de processar parte dos registros.

O orquestrador executa uma nova tentativa.

Se a escrita não for idempotente, podemos gerar duplicidades.

Conceitualmente:

Batch 100
   |
   v
Write
   |
   v
Failure
   |
   v
Retry Batch 100
   |
   v
Duplicate Records

Uma estratégia idempotente procura garantir que reprocessar a mesma unidade lógica de trabalho não produza efeitos adicionais indevidos.

Isso pode envolver:

• Chaves de negócio.
• Identificadores de batch.
• Operações MERGE.
• Deduplicação.
• Controle de commits.
• Escritas transacionais.
• Checkpoints.

A implementação precisa considerar o destino e a semântica de entrega.

Não basta utilizar dropDuplicates() indiscriminadamente.

---

29. Data Quality em streaming

Em pipelines streaming, a qualidade precisa ser avaliada continuamente.

Imagine:

Kafka
  |
  v
Spark Structured Streaming
  |
  v
Validation
  |
  +-- Valid Events
  |
  +-- Invalid Events
  |
  v
Lakehouse

Existem desafios adicionais:

• Eventos atrasados.
• Eventos fora de ordem.
• Duplicidades.
• Schema Evolution.
• Janelas temporais.
• Estado.
• Checkpoints.
• Reprocessamento.

Por exemplo, uma regra de unicidade em streaming pode exigir manutenção de estado e políticas de retenção.

Não podemos simplesmente aplicar uma janela global ilimitada sobre um fluxo infinito e esperar comportamento eficiente.

Esses detalhes serão aprofundados quando chegarmos a Structured Streaming.

---

30. Frameworks de Data Quality

Nem toda validação precisa ser implementada manualmente.

Existem ferramentas e bibliotecas que ajudam a estruturar regras, métricas e relatórios.

Entre as opções conhecidas estão:

Great Expectations

Framework de validação e documentação de expectativas sobre dados, com integrações que dependem da versão e do ambiente.

Deequ

Biblioteca voltada à definição e análise de restrições de qualidade em datasets, construída sobre Spark.

Soda

Ecossistema de ferramentas para verificações e monitoramento de qualidade de dados.

PySpark nativo

Permite implementar validações diretamente com DataFrames, Spark SQL e testes automatizados.

A escolha depende de:

• Integração com a plataforma.
• Volume.
• Linguagem.
• Requisitos de observabilidade.
• Manutenção.
• Licenciamento.
• Complexidade operacional.

Em alguns projetos, um conjunto pequeno de regras Spark bem testadas pode ser suficiente.

Em outros, uma plataforma dedicada traz vantagens importantes.

---

31. Arquitetura de referência

Vamos reunir os conceitos.

                      DATA PRODUCERS
                            |
                            v
                      DATA CONTRACTS
                            |
                            v
                       INGESTION
                            |
                            v
                        RAW LAYER
                            |
                            v
                    SPARK VALIDATION
                            |
                +-----------+-----------+
                |                       |
                v                       v
           VALID DATA              INVALID DATA
                |                       |
                v                       v
          LAKEHOUSE TABLE           QUARANTINE
                |
                v
          QUALITY METRICS
                |
                v
          DATA OBSERVABILITY
                |
                v
           DATA CATALOG
                |
                v
             LINEAGE
                |
                v
           DATA CONSUMERS

Na prática, observabilidade, catálogo, lineage e governança são capacidades transversais, não necessariamente etapas sequenciais do processamento.

Essa arquitetura ajuda a visualizar responsabilidades.

---

32. Quality Gates e publicação transacional

Uma estratégia robusta separa o processamento da publicação.

Conceitualmente:

1. Ingest Data
       |
       v
2. Transform
       |
       v
3. Validate
       |
       v
4. Calculate Quality Metrics
       |
       v
5. Evaluate Quality Gates
       |
       +-- FAIL --> Quarantine / Alert
       |
       v
6. Commit / Publish
       |
       v
7. Notify Consumers

O objetivo é impedir que dados reprovados sejam disponibilizados inadvertidamente.

Entretanto, a atomicidade entre múltiplas tabelas ou sistemas não deve ser presumida.

Se uma publicação envolver vários destinos, precisaremos definir explicitamente a estratégia de consistência e recuperação.

---

33. Indicadores para um produto de dados

Além das métricas técnicas do Spark, considero importante acompanhar indicadores de qualidade e confiabilidade.

Indicador	Objetivo
Valid Record Ratio	Proporção de registros aprovados
Null Rate	Completude de campos
Duplicate Rate	Ocorrência de duplicidades
Freshness	Atualidade do dataset
Schema Compatibility	Compatibilidade estrutural
Volume Anomaly	Variações inesperadas
Quality Gate Failures	Reprovações de publicação
Time to Detection	Tempo até detectar um problema
Time to Recovery	Tempo até recuperar o serviço
Contract Violations	Descumprimentos do contrato

Esses indicadores devem possuir definições claras.

Por exemplo, o denominador de uma taxa de duplicidade precisa ser documentado.

Sem definições consistentes, dashboards de qualidade podem gerar interpretações equivocadas.

---

34. Erros comuns

Alguns problemas aparecem repetidamente em plataformas de dados.

Validar apenas o schema.

Tipos corretos não garantem valores corretos.

Descartar registros inválidos silenciosamente.

Isso compromete rastreabilidade e pode ocultar perdas.

Utilizar um único Quality Score para tudo.

Problemas críticos podem ficar escondidos em uma média.

Não versionar contratos.

Mudanças incompatíveis podem quebrar consumidores.

Não monitorar freshness.

Um dataset estruturalmente correto pode estar desatualizado.

Não definir ownership.

Incidentes ficam sem responsáveis claros.

Confundir lineage técnico com lineage corporativo.

A rastreabilidade precisa abranger fontes, transformações e consumidores.

Acreditar que ACID resolve qualidade.

Transações consistentes podem armazenar dados semanticamente incorretos.

---

35. Como eu estruturaria a governança de um pipeline Spark

Para um projeto corporativo, começaria com um conjunto mínimo de controles:

1. Identificação do dataset e de seus responsáveis.
2. Contrato versionado.
3. Schema explícito.
4. Regras críticas de qualidade.
5. Métricas por execução.
6. Área de quarentena.
7. Quality Gates antes da publicação.
8. Logs e identificadores de execução.
9. Política de reprocessamento.
10. Monitoramento de freshness e volume.
11. Registro de lineage.
12. Catálogo e classificação.
13. Política de acesso e retenção.
14. Testes automatizados.
15. Procedimentos de resposta a incidentes.

O objetivo não é criar burocracia.

É tornar a confiabilidade dos dados verificável e operacional.

---

36. A principal lição deste capítulo

Até aqui, estudamos como fazer o Spark executar corretamente.

Agora precisamos ampliar nossa definição de sucesso.

Um pipeline confiável não é apenas aquele que:

Executes Successfully

Ele também precisa:

Respect Contracts
       |
       v
Validate Data
       |
       v
Track Quality
       |
       v
Preserve Lineage
       |
       v
Enforce Governance
       |
       v
Deliver Trusted Data

Essa é a diferença entre simplesmente processar dados e entregar um produto de dados confiável.

Performance torna o processamento eficiente. Qualidade torna os resultados utilizáveis. Governança torna o uso sustentável e responsável.

---

37. Como penso hoje

Quando avalio uma plataforma de dados, não pergunto apenas:

Quanto tempo o Spark demora para processar?

Também quero saber:

• Quem é responsável pelo dataset?
• Qual contrato define sua estrutura?
• Quais regras de qualidade são verificadas?
• Como identificamos registros inválidos?
• Como evitamos publicações incorretas?
• Como investigamos mudanças inesperadas?
• Como rastreamos a origem dos dados?
• Como sabemos se o dataset está atualizado?
• Como controlamos acesso e retenção?
• Como recuperamos uma execução com problemas?

Essas perguntas fazem parte da Engenharia de Dados tanto quanto CPU, memória e shuffle.

---

Próximo artigo

Agora que entendemos como validar e governar dados processados pelo Spark, vamos avançar para um dos temas mais importantes das arquiteturas modernas:

Apache Spark Structured Streaming.

Vamos explorar:

• Batch vs. Streaming.
• Microbatch.
• Structured Streaming.
• Sources e Sinks.
• Triggers.
• Checkpoints.
• Event Time vs. Processing Time.
• Watermarks.
• Stateful Processing.
• Exactly-once e suas limitações.
• Integração com Kafka.
• Escrita em Lakehouse.
• Observabilidade e recuperação.

A pergunta central será:

Como o Spark transforma fluxos contínuos de eventos em processamento distribuído confiável?

---

Série: Como aprendi Apache Spark

← Parte 13: Iceberg, Delta Lake e Hudi — a evolução para Lakehouse

Próximo: Parte 15 — Spark Structured Streaming: processando dados em tempo quase real