Calidad

`expect` conserva tanto filas válidas como inválidas y registra métricas; es útil durante adopción o para reglas informativas. `expect_or_drop` elimina las filas que incumplen y permite continuar. `expect_or_fail` detiene el flow y revierte atómicamente la actualización afectada cuando aceptar datos incorrectos sería peor que retrasar publicación.

La acción se elige por impacto y capacidad de remediación, no por severidad nominal. Un `order_id` nulo puede ir a cuarentena si el resto del pipeline debe continuar; una violación de unicidad en un saldo regulado puede justificar fallo. Toda regla necesita owner, definición y umbral operativo.

from pyspark import pipelines as dp

@dp.table(name="orders_silver")
@dp.expect("known_currency", "currency IN ('EUR', 'USD', 'GBP')")
@dp.expect_or_drop("valid_order_id", "order_id IS NOT NULL")
@dp.expect_or_fail("non_negative_amount", "amount >= 0")
def orders_silver():
    return spark.readStream.table("orders_bronze")

Descartar con una expectation no crea automáticamente una tabla de cuarentena. El patrón explícito clasifica una vista común en salida válida e inválida. La rama inválida añade `failure_reasons`, timestamp, source file u offset y versión de pipeline; la rama válida aplica las mismas condiciones complementarias.

La cuarentena necesita política de acceso porque puede contener PII, retención y un flujo de remediación. Reinyectar datos corregidos directamente en silver salta trazabilidad; conviene publicar una nueva entrada con identidad estable o una tabla de correcciones que pase por las mismas reglas.

from pyspark.sql import functions as F

classified = (
    spark.readStream.table("orders_bronze")
      .withColumn(
          "failure_reasons",
          F.array_compact(F.array(
              F.when(F.col("order_id").isNull(), F.lit("MISSING_ORDER_ID")),
              F.when(F.col("amount") < 0, F.lit("NEGATIVE_AMOUNT")),
              F.when(F.col("event_ts").isNull(), F.lit("INVALID_EVENT_TS")),
          ))
      )
)

valid = classified.where("size(failure_reasons) = 0")
quarantine = classified.where("size(failure_reasons) > 0")

Los eventos `flow_progress` incluyen estado, métricas y `data_quality` dentro de `details`. `origin` identifica pipeline, update y flow. La función `event_log(TABLE(...))` permite consultar el log asociado a una tabla del pipeline desde SQL y construir tendencias por expectation.

No todo campo interno es contrato público. Las consultas operativas seleccionan campos documentados y toleran ausencia de métricas en eventos que no sean de progreso. Conservar `update_id`, nombre de flow y timestamp permite relacionar una caída de calidad con despliegue y lote.

SELECT
  timestamp,
  origin.update_id AS update_id,
  origin.flow_name AS flow_name,
  details:flow_progress.status::STRING AS status,
  details:flow_progress.data_quality:expectations AS expectations
FROM event_log(TABLE(main.silver.orders_silver))
WHERE event_type = 'flow_progress'
ORDER BY timestamp DESC
LIMIT 100;

Validez, completitud, unicidad, consistencia, puntualidad y exactitud requieren evidencias diferentes. `amount >= 0` comprueba validez, pero no exactitud frente al sistema de pagos. Una expectation fila a fila tampoco demuestra unicidad global sin una transformación o agregación adecuada.

Las reglas evolucionan como código versionado. Un cambio de umbral debe revisar impacto histórico y despliegue; renombrar una expectation rompe series de métricas. Las reglas críticas pueden agruparse con `expect_all_or_fail`, mientras reglas de observación usan `expect_all` para una configuración compartida.

structural_rules = {
    "order_id_present": "order_id IS NOT NULL",
    "event_ts_present": "event_ts IS NOT NULL",
    "amount_non_negative": "amount >= 0",
}

business_observations = {
    "known_currency": "currency IN ('EUR', 'USD', 'GBP')",
    "reasonable_amount": "amount <= 100000",
}

Un conteo absoluto de inválidos confunde crecimiento de volumen con degradación. La tasa `failed / (passed + failed)` por expectation, flow y ventana permite comparar. Expectations observan, descartan o fallan registros dentro del pipeline: son enforcement vinculado al código y al update.

Data Quality Monitoring complementa ese contrato sin modificar las tablas monitorizadas ni añadir trabajo al job productor: anomaly detection aprende patrones históricos para freshness y completeness, y data profiling calcula estadísticas y drift. Se ejecuta en serverless con coste propio. Úsalo para descubrir degradaciones transversales; conserva expectations o reconciliaciones cuando una regla debe bloquear o aislar datos.

SELECT
  window_start,
  flow_name,
  expectation_name,
  failed_records,
  passed_records,
  failed_records / NULLIF(failed_records + passed_records, 0) AS failure_rate
FROM main.ops.pipeline_expectation_metrics
WHERE window_start >= current_timestamp() - INTERVAL 24 HOURS
ORDER BY failure_rate DESC;