Proyecto pipelines

Pedidos append y clientes CDC tienen naturalezas distintas: los primeros encajan en streaming tables; los segundos en AUTO CDC. Gold necesita métricas consistentes con updates de ambas fuentes y puede ser una materialized view. El diseño identifica owners y fronteras bronze/silver/gold con nombres gobernados.

Los NFR se convierten en decisiones comprobables: p95 de frescura, umbral de errores, RTO, retención y presupuesto. Un diagrama sin criterios de aceptación no permite saber si el pipeline está terminado ni qué hacer ante un fallo.

domain: commerce
sources:
  orders:
    semantics: append
    key: order_id
    freshness_p95_minutes: 10
  customers:
    semantics: cdc
    key: customer_id
    sequence: source_lsn
targets:
  silver_orders: main.silver.orders
  current_customers: main.silver.customers_current
  daily_sales: main.gold.daily_customer_sales
rto_minutes: 60
owner: commerce-data

Pedidos se ingieren en bronze y se validan en silver con lectura streaming. Clientes llegan como operaciones con `source_lsn`; AUTO CDC materializa el estado actual SCD 1 o el historial SCD 2. La streaming table target se declara antes del flow y las columnas operativas se excluyen.

Gold une pedidos con clientes según la necesidad temporal. Si se necesita el cliente actual, SCD 1 basta; si la segmentación al momento del pedido importa, se usa SCD 2 y un join por intervalo. Esa decisión cambia corrección, coste y complejidad.

CREATE OR REFRESH STREAMING TABLE main.silver.customers_current;

CREATE FLOW customers_current_cdc AS AUTO CDC INTO
  main.silver.customers_current
FROM STREAM(main.bronze.customers_cdc)
KEYS (customer_id)
APPLY AS DELETE WHEN operation = 'DELETE'
SEQUENCE BY source_lsn
COLUMNS * EXCEPT (operation)
STORED AS SCD TYPE 1;

Pedidos sin ID van a cuarentena con archivo y motivo. Monedas desconocidas pueden observarse mientras negocio decide. Un importe negativo en una tabla financiera puede fallar la actualización. La clasificación compartida evita que una fila desaparezca entre filtros inconsistentes.

El event log alimenta un dashboard por update y flow. La aceptación incluye casos para cada acción, una tasa máxima y un procedimiento de reentrada. Proteger la cuarentena con permisos más estrictos evita convertir observabilidad en fuga de PII.

@dp.table(name="orders_silver")
@dp.expect("known_currency", "currency IN ('EUR', 'USD', 'GBP')")
@dp.expect_or_fail("non_negative_amount", "amount >= 0")
def orders_silver():
    return (
        spark.readStream.table("orders_classified")
          .where("size(failure_reasons) = 0")
          .drop("failure_reasons")
    )

Una pipeline task actualiza datasets declarativos. Una tarea posterior consulta event log y reconciliación; una If/else bloquea publicación si se supera el umbral. Cleanup y notificación usan Run if. El mismo bundle define recursos para dev, test y prod con identidades y catálogos diferentes.

La promoción no consiste en copiar notebooks. Se valida el bundle, se despliega el mismo artefacto, se ejecuta un smoke test y se compara lineage/esquema. Producción se activa con trigger pausado inicialmente y rollback conocido.

tasks:
  - task_key: update_pipeline
    pipeline_task:
      pipeline_id: orders_pipeline
  - task_key: validate_update
    depends_on:
      - task_key: update_pipeline
    notebook_task:
      notebook_path: /Workspace/commerce/validate_update
  - task_key: quality_gate
    depends_on:
      - task_key: validate_update
    condition_task:
      left: "{{tasks.validate_update.values.failure_rate}}"
      op: LESS_THAN_OR_EQUAL
      right: "0.01"

El equipo prueba una segunda ejecución sin datos, un update CDC fuera de orden, una fila en cuarentena y una caída recuperable. Un backfill de una fecha usa el mismo pipeline/Job y se reconcilia. El event log debe explicar qué flows procesaron filas y qué expectations fallaron.

El runbook identifica owner, SLO, paneles, decisiones de retry, repair o full refresh y rutas de rollback. La entrega incluye consultas de evidencia y límites conocidos. Una demo feliz sin incidente ni recuperación no valida producción.

acceptance:
  - second_run_adds_zero_duplicates
  - out_of_order_cdc_keeps_highest_sequence
  - invalid_order_is_quarantined_with_reason
  - expectation_metrics_visible_in_event_log
  - one_day_backfill_reconciles_to_manifest
  - repair_run_preserves_successful_outputs
  - rto_under_60_minutes
  - service_principal_has_least_privilege