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:

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 p...