Skip to content

dead_letter_pattern.md

Latest commit

 

History

History
478 lines (320 loc) · 33.5 KB

dead_letter_pattern.md

File metadata and controls

478 lines (320 loc) · 33.5 KB
Raw

Dead Letter Pattern en Pipelines de Datos

Resiliencia ante Registros Corruptos en la Ingesta de Datos de Contratación Pública con Apache Spark y Delta Lake

En la construcción de un Data Lake robusto sobre fuentes de datos externas, uno de los retos más frecuentes y menos visibles es la gestión de los registros que no se pueden procesar correctamente. Una estrategia ingenua consiste en abortar el pipeline completo ante el primer error, lo cual hace que un único fichero mal formado bloquee la ingesta de miles de registros válidos. Una estrategia igualmente peligrosa es simplemente descartar los registros fallidos sin dejar rastro, lo que compromete la auditabilidad y la capacidad de recuperación.

El Dead Letter Pattern (Patrón de Mensajes Fallidos) es la solución estándar de la industria para este problema. Este documento explica qué es, por qué es necesario en nuestro proyecto específico, y detalla con precisión técnica cómo ha sido implementado en la capa Silver del pipeline de la Plataforma de Contratación del Sector Público (PLACSP).

1. ¿Qué es el Dead Letter Pattern?

1.1 Origen del concepto

El término Dead Letter proviene del mundo postal: una "carta muerta" (dead letter) es aquella que no puede ser entregada a su destinatario ni devuelta al remitente. Las oficinas postales no destruían estas cartas; las enviaban a una Oficina de Cartas Muertas (Dead Letter Office) donde quedaban almacenadas para su revisión y posible recuperación posterior.

En ingeniería de software, este mismo principio fue adoptado por los sistemas de mensajería como IBM MQ, RabbitMQ y Apache Kafka, donde las colas o tópicos especiales llamados Dead Letter Queues (DLQ) almacenan los mensajes que no pudieron ser procesados. Más tarde, el patrón fue adoptado por la ingeniería de datos como mecanismo de gestión de errores a nivel de registro en pipelines ETL/ELT.

1.2 Definición formal

El Dead Letter Pattern es una estrategia de manejo de errores que establece que, cuando un registro no puede ser procesado correctamente durante un pipeline de datos (ya sea por corrupción, incompatibilidad de esquema, valores inválidos u otros errores), dicho registro no debe descartarse silenciosamente ni bloquear el procesamiento del resto. En su lugar, debe ser:

  1. Redirigido a una ubicación de almacenamiento separada y dedicada (la dead letter store).
  2. Enriquecido con metadatos sobre el error: qué falló, cuándo, en qué fichero, en qué etapa del pipeline.
  3. Preservado de forma duradera para permitir su revisión, diagnóstico y reprocesamiento posterior.

1.3 Las tres propiedades fundamentales

Un Dead Letter Store correctamente implementado debe satisfacer tres propiedades:

Propiedad 1: No bloqueo (Non-blocking)

El pipeline principal debe continuar procesando el resto de los registros válidos con total normalidad. La presencia de un registro corrupto no debe interrumpir ni degradar el procesamiento del batch. Esta propiedad es la más crítica, pues es la que diferencia el patrón de un simple bloque try/except que aborta la ejecución.

Propiedad 2: Trazabilidad (Traceability)

Cada registro fallido almacenado en el dead letter store debe ir acompañado de suficiente información para responder las siguientes preguntas:

  • ¿De qué fichero o fuente proviene el registro?
  • ¿En qué etapa del pipeline ocurrió el fallo?
  • ¿Cuál es la naturaleza del error (tipo de excepción, mensaje)?
  • ¿Cuándo ocurrió el fallo?

Sin esta información, la dead letter store se convierte en un vertedero de datos inaccesible.

Propiedad 3: Replayabilidad (Replayability)

El dead letter store debe estar diseñado de tal manera que los registros fallidos puedan ser reprocesados en el futuro, una vez que se haya corregido el problema que causó el fallo. Esto implica preservar suficiente contexto del registro original para que pueda volver a entrar en el pipeline.

2. El problema que resuelve en pipelines de datos

Un pipeline de procesamiento batch que lee un conjunto de ficheros y los transforma tiene, por defecto, un comportamiento todo-o-nada: o todos los ficheros se procesan correctamente o la ejecución falla. Este comportamiento proviene del modelo de ejecución de Apache Spark, donde un error en cualquier tarea (task) de un job puede propagar la excepción hasta el driver y detener la ejecución completa. Esto crea un problema especialmente severo en pipelines que procesan datos de fuentes externas, donde la calidad de los datos no está bajo nuestro control directo.

El extremo opuesto es igualmente problemático. Algunas configuraciones de lectura (como el modo DROPMALFORMED en Spark) descartan silenciosamente los registros que no se pueden parsear. Aunque el pipeline termina sin errores, el resultado es una pérdida de datos sin registro: no sabemos cuántos registros se perdieron, por qué, ni de qué ficheros provenían. En el contexto de datos de contratación pública, esto podría implicar que licitaciones adjudicadas queden ausentes del análisis sin que nadie lo detecte.

El Dead Letter Pattern es la única estrategia que satisface simultáneamente las dos necesidades: continuidad del pipeline y trazabilidad completa de los errores. La siguiente tabla resume el comportamiento de cada estrategia disponible en Spark:

Estrategia Comportamiento ante error Impacto en el pipeline Trazabilidad
FAILFAST Lanza excepción, aborta todo Pipeline bloqueado Alta (el error es visible)
DROPMALFORMED Descarta el registro Pipeline continúa Nula (pérdida silenciosa)
PERMISSIVE + Dead Letter Captura el registro y lo desvía Pipeline continúa Alta (registro completo del fallo)

3. Por qué lo hemos implementado en este proyecto

3.1 Características de la fuente de datos

El pipeline PLACSP ingiere datos de la Plataforma de Contratación del Sector Público, una fuente de datos externa mantenida por el gobierno español sobre la que no tenemos ningún control de calidad. Las características de esta fuente hacen que el Dead Letter Pattern no sea un lujo, sino una necesidad:

  • Heterogeneidad del estándar CODICE: Los ficheros ATOM/XML siguen el estándar CODICE, pero la interpretación y el cumplimiento de dicho estándar varía entre los diferentes organismos que publican licitaciones. Entidades como diputaciones, ministerios o empresas públicas generan ficheros que ocasionalmente tienen estructuras inesperadas, campos faltantes, o incluso XML malformado.

  • Volumen: El pipeline histórico procesa potencialmente cientos de ficheros .atom por año y por dataset (hasta 5 datasets). Un único fichero corrupto en el dataset licitaciones del año 2023, sin Dead Letter, bloquearía la ingesta de todos los demás ficheros del mismo año.

  • Operación autónoma: El script de extracción incremental (daily_extraction.py) está diseñado para ejecutarse de forma autónoma y desatendida como parte de una orquestación (Apache Airflow). En este contexto, no hay un operador humano monitorizando cada ejecución, por lo que la resiliencia automática ante errores es crítica.

  • Idemponcia: El pipeline implementa operaciones MERGE sobre Delta Lake, lo que significa que está diseñado para ser re-ejecutable. El Dead Letter Pattern complementa perfectamente esta idempotencia: los registros fallidos pueden ser reprocesados de forma aislada sin volver a ejecutar el pipeline completo.

3.2 El punto concreto de fallo y su evolución

El análisis del código reveló que el punto de mayor fragilidad era la lectura y parseo de los ficheros XML/ATOM. La implementación ha evolucionado en dos etapas para garantizar la captura efectiva de todos los registros corruptos.

Implementación inicial: La lectura usaba rowTag="feed" con un esquema raíz (root_schema) que capturaba el documento ATOM completo como una fila. Esto significa que la unidad de fallo era el fichero entero: si cualquier aspecto del <feed> no podía parsearse, el feed completo se marcaba como corrupto. Aunque funcional para XML estrictamente malformado, este enfoque tenía dos limitaciones:

  1. La granularidad era demasiado gruesa: un error en un único <entry> podía comprometer la captura de todos los demás entries del mismo fichero.
  2. La opción columnNameOfCorruptRecord presentaba un comportamiento no documentado con mismatches estructurales que causaba pérdida silenciosa de datos (ver sección 4.3).

Implementación actual (dual-read): Se sustituyó el esquema root_schema por dos lecturas independientes y paralelas a nivel de elemento: una con rowTag="entry" y bidding_schema, y otra con rowTag="at:deleted-entry" y deleted_schema. Esta refactorización, junto con la corrección de los mismatches estructurales del esquema, garantiza que la unidad de fallo sea el <entry> individual y que el mecanismo columnNameOfCorruptRecord opere de forma correcta y predecible.

4. Cómo está implementado

La implementación sigue el principio de mínima invasión: se integra en el punto más temprano del pipeline (la lectura del fichero) y utiliza mecanismos nativos de Apache Spark para no introducir overhead adicional.

4.1 Arquitectura de la solución

El flujo de datos comienza con dos lecturas en paralelo de todos los ficheros ATOM del dataset mediante un glob pattern. Cada lectura parsea una capa diferente del documento ATOM: los <entry> con licitaciones activas y los <at:deleted-entry> con licitaciones eliminadas. En cada lectura, Spark separa los registros en dos categorías: aquellos que pudieron ser parseados correctamente contra el esquema definido, y aquellos que no. Los registros válidos continúan por el pipeline normal hacia las tablas Silver (entries, lots, documents, results). Los registros corruptos se desvían inmediatamente, antes de cualquier transformación, hacia la tabla Delta de dead letters ubicada en silver/data/dead_letters/, particionada por nombre de dataset.

flowchart TD
    A[("Ficheros ATOM (Bronze)")] --> B1 & B2

    B1["spark.read (rowTag='entry', bidding_schema)"]
    B2["spark.read (rowTag='at:deleted-entry', deleted_schema)"]

    B1 --> C1{_corrupt_record IS NULL?}
    B2 --> C2{_corrupt_record IS NULL?}

    C1 -- "Sí (válido)" --> D1["df_entries.drop(_corrupt_record)"]
    C1 -- "No (corrupto)" --> E1["df_corrupt_entries"]
    C2 -- "Sí (válido)" --> D2["df_deleted.drop(_corrupt_record)"]
    C2 -- "No (corrupto)" --> E2["df_corrupt_deleted"]

    D1 & D2 --> F["Extracción"]
    F --> G[("Silver Tables (entries, lots, documents, results)")]

    E1 & E2 --> H["write_dead_letter()"]
    H --> I[("silver/data/dead_letters/")]
Loading

4.2 El mecanismo central: modo PERMISSIVE de spark-xml

La piedra angular del patrón es la combinación de dos opciones de lectura en el conector spark-xml (com.databricks:spark-xml):

1. mode = "PERMISSIVE"

Cuando el modo de lectura es PERMISSIVE, el comportamiento ante un registro XML que no puede ser parseado correctamente no es lanzar una excepción, sino intentar parsear lo que pueda y registrar el fallo. Esto contrasta con los otros dos modos disponibles, FAILFAST y DROPMALFORMED.

2. columnNameOfCorruptRecord = "_corrupt_record"

Esta opción indica a Spark qué columna del esquema de destino debe usar para almacenar el contenido del registro que no pudo ser parseado. Para que funcione, es obligatorio que dicha columna exista en el esquema de lectura definido. Tanto bidding_schema como deleted_schema en silver/src/schema.py incluyen este campo como último StructField:

# En bidding_schema y deleted_schema:
StructField("_corrupt_record", StringType(), True)

Cuando spark-xml encuentra un <entry> o <at:deleted-entry> que no puede mapear al esquema, coloca su contenido XML en bruto dentro de _corrupt_record. Para los registros válidos, este campo es NULL.

Advertencia importante: Como se detalla en la sección 4.3, la opción columnNameOfCorruptRecord tiene un comportamiento no documentado con mismatches estructurales de tipo que puede causar pérdida silenciosa de datos. Es imprescindible que el esquema definido coincida fielmente con la estructura real del XML.

3. El requisito de nullable=True en todos los campos del esquema

Para que el mecanismo PERMISSIVE funcione correctamente con columnNameOfCorruptRecord, todos los campos del esquema deben declararse con nullable=True, incluyendo campos que conceptualmente nunca tendrán valores nulos en los datos válidos (como id, title o updated).

La razón es una sutileza del modelo de ejecución de Spark. Cuando spark-xml procesa una entrada corrupta en modo PERMISSIVE, construye un objeto Row donde _corrupt_record contiene el XML en bruto y todos los demás campos del esquema son null. Este Row corrupto existe en el DataFrame antes de que el filtro col("_corrupt_record").isNull() pueda eliminarlo.

El problema surge porque el encoder de Spark —que serializa las filas para su transmisión entre stages o para operaciones de escritura— valida la restricción nullable=False en el momento de serializar cada fila, no después del filtro. Con campos marcados como nullable=False, el encoder genera una instrucción assertnotnull que falla al intentar codificar la fila corrupta (con todos esos campos a null), lanzando:

SparkRuntimeException: Error while encoding:
  java.lang.RuntimeException: The Nth field 'X' of input row cannot be null.

Este error ocurre aunque el DataFrame filtrado final no contenga ningún null, porque el encoder se ejecuta antes de que el filtro tenga oportunidad de actuar.

El nullable=True no implica que los datos válidos vayan a tener nulos en esos campos. Los entries Atom correctamente formados siempre tienen id, title, updated, etc. (garantizado por el RFC 4287). La marca nullable=True es un requisito del mecanismo de parsing, no una descripción del dominio de negocio. La integridad de los datos válidos la garantiza el filtro _corrupt_record.isNull(), no la restricción de nullabilidad del esquema de lectura.

4.3 El problema de los mismatches estructurales con columnNameOfCorruptRecord

Esta sección documenta un comportamiento crítico y no documentado de spark-xml que fue descubierto durante el desarrollo del pipeline y que motivó la corrección del esquema bidding_schema.

El síntoma

Al habilitar columnNameOfCorruptRecord en la lectura con rowTag="entry" y bidding_schema, el pipeline extraía únicamente 12 filas donde se esperaban miles. Sin la opción habilitada (modo PERMISSIVE sin columnNameOfCorruptRecord), la extracción funcionaba correctamente y retornaba el número esperado de registros.

La causa raíz

El comportamiento de columnNameOfCorruptRecord ante los diferentes tipos de discrepancia entre el esquema definido y la estructura real del XML es el siguiente:

Tipo de discrepancia Comportamiento
XML malformado (bytes corruptos) Enruta correctamente a _corrupt_record
Valor de tipo incompatible (ej. string donde se espera int) Enruta correctamente a _corrupt_record
Schema define ArrayType, XML tiene elemento único (StructType) Descarta silenciosamente el registro
Schema define StructType, XML tiene múltiples elementos (ArrayType) Descarta silenciosamente el registro

La discrepancia del tercer caso era la causa del problema. Cuando el modo es PERMISSIVE sin columnNameOfCorruptRecord, spark-xml envuelve automáticamente el elemento único en un array de un elemento (comportamiento tolerante). Pero cuando columnNameOfCorruptRecord está activo, spark-xml aplica validación estricta de estructura y, ante cualquier mismatch estructural, descarta el registro sin rutarlo a _corrupt_record ni emitir ninguna advertencia.

Los campos afectados

La comparación entre el esquema definido manualmente (bidding_schema) y el esquema inferido por spark-xml sobre datos reales reveló tres campos con esta discrepancia:

Campo Schema definido (anterior) Schema inferido / real
location_schema.cac:AddressLine ArrayType(StructType([cbc:Line])) StructType([cbc:Line])
term_schema.cac:Language ArrayType(StructType([cbc:ID])) StructType([cbc:ID])
document_schema.cac:Attachment ArrayType(StructType([cac:ExternalReference])) StructType([cac:ExternalReference])

Los tres campos habían sido definidos originalmente como ArrayType por precaución ante la posibilidad de múltiples valores. Sin embargo, el estándar CODICE y los datos reales demuestran que estos elementos son siempre singulares y por tanto se cambiaron a StructType.

4.4 La lectura dual por rowTag

Del enfoque de feed al enfoque de entry

La implementación original leía los ficheros ATOM con rowTag="feed", produciendo una fila por fichero. Esta fila contenía el documento ATOM completo, con los <entry> anidados en un array dentro del esquema root_schema. El pipeline posteriormente hacía explode() sobre ese array para obtener una fila por entry.

La implementación actual lee directamente con rowTag="entry", produciendo una fila por elemento <entry> del documento ATOM. De forma análoga, rowTag="at:deleted-entry" produce una fila por elemento de entrada eliminada.

# Lectura de licitaciones activas
df_entries = spark.read \
    .format("xml") \
    .option("rowTag", "entry") \
    .schema(bidding_schema) \
    .option("mode", "PERMISSIVE") \
    .option("columnNameOfCorruptRecord", "_corrupt_record") \
    .option("nullValue", "") \
    .load(atom_glob)

# Lectura de licitaciones eliminadas
df_deleted = spark.read \
    .format("xml") \
    .option("rowTag", "at:deleted-entry") \
    .schema(deleted_schema) \
    .option("mode", "PERMISSIVE") \
    .option("columnNameOfCorruptRecord", "_corrupt_record") \
    .option("nullValue", "") \
    .load(atom_glob)

Ventajas del enfoque dual

  1. Granularidad de fallo más fina: La unidad de fallo es ahora el <entry> individual (no el fichero entero). Un entry corrupto no afecta a los demás entries del mismo fichero.

  2. Schemas independientes y más simples: bidding_schema y deleted_schema son más simples y directos que el antiguo root_schema, lo que reduce la superficie de posibles mismatches y facilita el mantenimiento.

  3. Eliminación del explode(): Las funciones de extracción trabajan directamente sobre el DataFrame resultante sin necesidad de aplanar arrays. Esto simplifica el código y evita un paso de transformación potencialmente costoso.

  4. _corrupt_record en el schema correcto: Cada schema define su propio campo _corrupt_record, lo que hace el contrato explícito y fácil de razonar.

input_file_name() y la columna _source_file

Con el enfoque dual de rowTag="entry", spark-xml produce una fila por elemento XML individual (no por fichero). La función input_file_name() puede igualmente retornar el path correcto del fichero que contiene cada entry, por lo que se añade explícitamente mediante .withColumn("_source_file", input_file_name()) justo después de cada lectura, antes del split válido/corrupto. La columna _source_file se usa en write_dead_letter() para registrar de qué fichero proviene cada entrada corrupta, y se elimina del DataFrame de registros válidos antes de pasarlo a las funciones de extracción (.drop("_corrupt_record", "_source_file")). Este patrón está presente en ambos scripts: extract_biddings.py y daily_extraction.py.

4.5 El split valid/corrupt

Inmediatamente después de cada lectura, se realiza un split del DataFrame en dos subconjuntos mutuamente excluyentes:

df_corrupt_entries = df_entries.filter(col("_corrupt_record").isNotNull())
df_corrupt_deleted = df_deleted.filter(col("_corrupt_record").isNotNull())

write_dead_letter(df_corrupt_entries, dead_letter_path, dataset_name, "xml_parsing")
write_dead_letter(df_corrupt_deleted, dead_letter_path, dataset_name, "xml_parsing")

df_entries = df_entries.filter(col("_corrupt_record").isNull()).drop("_corrupt_record")
df_deleted = df_deleted.filter(col("_corrupt_record").isNull()).drop("_corrupt_record")

Este split es lazy en Spark: no se ejecuta en este momento, sino que se registra como un par de nodos en el plan de ejecución. Spark leerá cada fichero una única vez y dividirá el resultado en ambas ramas del DAG en la misma pasada. No hay lectura doble de los ficheros.

El drop de _corrupt_record antes de pasar los DataFrames al pipeline de extracción es necesario porque esta columna no forma parte del esquema esperado por las funciones de extracción (extract_bidding_entries, extract_lots, etc.) y causaría errores aguas abajo.

4.6 El módulo dead_letter.py

Toda la lógica de escritura está encapsulada en el módulo silver/src/dead_letter.py, que expone una única función pública:

def write_dead_letter(
    df_corrupt: DataFrame,
    dead_letter_path: str,
    dataset: str,
    stage: str,
    error_type: str = "CORRUPT_RECORD",
) -> int:

Parámetros

  • df_corrupt: DataFrame que contiene, como mínimo, las columnas _source_file (path del fichero origen, añadida mediante input_file_name() justo después de cada lectura XML) y _corrupt_record (contenido del registro que falló).
  • dead_letter_path: Path absoluto de la tabla Delta de dead letters. Por convención del proyecto, se ubica en silver/data/dead_letters/.
  • dataset: Nombre del dataset que estaba siendo procesado cuando ocurrió el fallo (p.ej. licitaciones, menores, agregadas). Se usa como valor de la columna de partición.
  • stage: Etapa del pipeline donde ocurrió el fallo. Actualmente se usa "xml_parsing", pero el diseño permite extenderlo en el futuro a otras etapas como "schema_validation" o "merge".
  • error_type: Tipo de error, por defecto "CORRUPT_RECORD". Puede tomar otros valores para distinguir entre distintas clases de fallos.

Lógica interna

La función realiza las siguientes operaciones:

  1. Guarda de cortocircuito: Si el DataFrame no tiene registros (count() == 0), retorna 0 sin realizar ninguna operación de escritura. Esto evita crear transacciones vacías en Delta Lake.

  2. Log de advertencia: Emite un mensaje de nivel WARNING con el número de registros fallidos, el dataset y la ubicación del dead letter store. Esto asegura que los fallos sean visibles en los logs del pipeline, aunque éste continúe ejecutándose.

  3. Construcción del DataFrame de dead letters: Selecciona y renombra las columnas necesarias, y añade los metadatos contextuales mediante lit() y current_timestamp():

df_dead = df_corrupt.select(
    col("_source_file").alias("source_file"),         # Path del fichero ATOM
    lit(dataset).alias("dataset"),                    # Nombre del dataset
    lit(stage).alias("stage"),                        # Etapa del pipeline
    lit(error_type).alias("error_type"),              # Tipo de error
    col("_corrupt_record").alias("error_message"),    # Contenido XML del registro fallido
    current_timestamp().alias("processing_timestamp") # Momento exacto del fallo
)
  1. Escritura en Delta Lake: Escribe el DataFrame en modo append (nunca sobreescribe) con particionamiento por dataset:
df_dead.write \
    .format("delta") \
    .mode("append") \
    .partitionBy("dataset") \
    .save(dead_letter_path)

Schema de la tabla Dead Letter

Columna Tipo Descripción
source_file String Path absoluto del fichero ATOM que contenía el registro corrupto
dataset String Nombre del dataset PLACSP (p.ej. licitaciones, menores)
stage String Etapa del pipeline donde ocurrió el fallo (xml_parsing)
error_type String Tipo de error (CORRUPT_RECORD)
error_message String Contenido XML en bruto del registro que no pudo ser parseado
processing_timestamp Timestamp Momento en que el registro fue escrito en el dead letter store

4.7 Integración en transform_biddings.py

El script de carga histórica por año fue modificado en la función process_dataset. La lectura dual reemplaza a la lectura única anterior:

from pyspark.sql.functions import col, input_file_name
from bronze.src import bidding_schema, deleted_schema, write_dead_letter

atom_glob = f"file://{raw_dataset_path}/*.atom"

df_entries = spark.read \
    .format("xml") \
    .option("rowTag", "entry") \
    .schema(bidding_schema) \
    .option("mode", "PERMISSIVE") \
    .option("columnNameOfCorruptRecord", "_corrupt_record") \
    .option("nullValue", "") \
    .load(atom_glob) \
    .withColumn("_source_file", input_file_name())

df_deleted = spark.read \
    .format("xml") \
    .option("rowTag", "at:deleted-entry") \
    .schema(deleted_schema) \
    .option("mode", "PERMISSIVE") \
    .option("columnNameOfCorruptRecord", "_corrupt_record") \
    .option("nullValue", "") \
    .load(atom_glob) \
    .withColumn("_source_file", input_file_name())

dead_letter_path = str(bronze_data_path / "dead_letters")
df_corrupt_entries = df_entries.filter(col("_corrupt_record").isNotNull())
df_corrupt_deleted = df_deleted.filter(col("_corrupt_record").isNotNull())
write_dead_letter(df_corrupt_entries, dead_letter_path, dataset_name, "xml_parsing")
write_dead_letter(df_corrupt_deleted, dead_letter_path, dataset_name, "xml_parsing")

df_entries = df_entries.filter(col("_corrupt_record").isNull()).drop("_corrupt_record", "_source_file")
df_deleted = df_deleted.filter(col("_corrupt_record").isNull()).drop("_corrupt_record", "_source_file")

# Se cachean para reusar en las múltiples extracciones sin releer el XML
df_entries.cache()
df_deleted.cache()

Este patrón garantiza que incluso si varios ficheros de un año tienen entries malformados, el resto se procesa correctamente y los fallidos quedan documentados.

4.8 Integración en daily_extraction.py

El script de ingesta incremental usa Structured Streaming con foreachBatch. El Dead Letter Pattern se integra dentro del closure process_micro_batch, que es invocado por Spark para cada micro-batch de ficheros nuevos:

def _make_batch_processor(spark, dataset_name, output_path, parties_path, silver_data_path, df_catalog):

    dead_letter_path = str(silver_data_path / "dead_letters")  # calculado una vez en el closure

    def process_micro_batch(df_batch, batch_id: int) -> None:
        new_files = [row.path for row in df_batch.select("path").collect()]
        file_list = ",".join(new_files)

        df_xml_entries = spark.read \
            .format("xml") \
            .option("rowTag", "entry") \
            .schema(bidding_schema) \
            .option("mode", "PERMISSIVE") \
            .option("columnNameOfCorruptRecord", "_corrupt_record") \
            .option("nullValue", "") \
            .load(file_list)

        df_xml_deleted = spark.read \
            .format("xml") \
            .option("rowTag", "at:deleted-entry") \
            .schema(deleted_schema) \
            .option("mode", "PERMISSIVE") \
            .option("columnNameOfCorruptRecord", "_corrupt_record") \
            .option("nullValue", "") \
            .load(file_list)

        df_corrupt_entries = df_xml_entries.filter(col("_corrupt_record").isNotNull())
        df_corrupt_deleted = df_xml_deleted.filter(col("_corrupt_record").isNotNull())
        write_dead_letter(df_corrupt_entries, dead_letter_path, dataset_name, "xml_parsing")
        write_dead_letter(df_corrupt_deleted, dead_letter_path, dataset_name, "xml_parsing")

        df_xml_entries = df_xml_entries.filter(col("_corrupt_record").isNull()).drop("_corrupt_record")
        df_xml_deleted = df_xml_deleted.filter(col("_corrupt_record").isNull()).drop("_corrupt_record")

        # Pipeline normal continúa con df_xml_entries y df_xml_deleted válidos...

El dead_letter_path se calcula una sola vez en el momento de crear el closure (no en cada invocación del micro-batch), lo que evita una operación de construcción de strings repetida en cada batch.

5. Consideraciones de diseño

5.1 Por qué Delta Lake para el dead letter store

La elección de Delta Lake como formato de almacenamiento del dead letter store frente a alternativas como Parquet plano, JSON, o CSV es deliberada:

  • ACID Transactions: Las escrituras en Delta son atómicas. Si el proceso de escritura del dead letter falla a mitad (por ejemplo, por un error de red), no se generan ficheros parciales que corrompan el estado de la tabla.
  • Append-only con historial: Delta mantiene un log de transacciones que permite ver qué registros fallidos fueron añadidos en cada ejecución, facilitando la auditoría.
  • Consultas SQL: Al estar en Delta, la tabla de dead letters puede ser consultada directamente con Spark SQL o DuckDB como cualquier otra tabla del Data Lake, sin necesidad de herramientas especiales.
  • Consistencia con el resto del Data Lake: Todas las tablas de la capa Silver son Delta. Mantener el mismo formato en el dead letter store simplifica la arquitectura.

5.2 Por qué el particionamiento por dataset

La tabla de dead letters está particionada por la columna dataset. Esta decisión se justifica por el patrón de acceso esperado: cuando un operador quiera revisar o reprocesar los registros fallidos, lo hará dataset a dataset (revisará los fallos de licitaciones de forma independiente a los de menores). El particionamiento asegura que Spark solo lea los ficheros del dataset relevante, sin escanear toda la tabla.

5.3 Lo que el patrón NO cubre actualmente

La implementación cubre específicamente los fallos en la etapa de parseo XML (la más común y crítica). Existen otras etapas del pipeline donde pueden ocurrir fallos que actualmente no están capturados por el Dead Letter Pattern:

  • Fallos en las transformaciones de extracción (extract_bidding_entries, extract_lots, etc.): Errores en expresiones Spark SQL como expr() o transform() sobre campos de esquema incorrecto.
  • Fallos en las operaciones MERGE de Delta: Incompatibilidades de esquema entre el DataFrame de origen y la tabla Delta de destino.
  • Fallos en la carga del catálogo de genericodes: Si el fichero gc_codes.parquet no está disponible.

Cubrir estas etapas adicionales requeriría un esfuerzo mayor y está previsto como una evolución futura del patrón, potencialmente coordinada con la implementación de la orquestación en Apache Airflow, que dispondrá de sus propios mecanismos de retry y alertas a nivel de tarea.

6. Cómo operar con el dead letter store

Una vez que el pipeline ha procesado datos, los registros fallidos pueden ser consultados directamente con Spark o con cualquier motor compatible con Delta Lake (DuckDB, etc.):

# Ver todos los registros fallidos del dataset 'licitaciones'
df_dead = spark.read.format("delta").load("silver/data/dead_letters/")
df_dead.filter(col("dataset") == "licitaciones").show(truncate=False)
-- Resumen de fallos por dataset y fecha
SELECT
    dataset,
    DATE(processing_timestamp) AS fecha,
    COUNT(*) AS num_fallos,
    COUNT(DISTINCT source_file) AS ficheros_afectados
FROM delta.`silver/data/dead_letters/`
GROUP BY ALL
ORDER BY fecha DESC, num_fallos DESC;

7. Evolución futura

El Dead Letter Pattern implementado es la versión mínima viable necesaria para el estado actual del proyecto. A medida que el pipeline madure, se contempla la siguiente evolución:

Fase Mejora prevista
Actual Dead letter en etapa de parseo XML (xml_parsing) con lectura dual por rowTag e input_file_name() para trazabilidad de fichero
Con Airflow Alertas automáticas cuando el volumen de dead letters supera un umbral
Con Airflow Tareas de reprocesamiento automático de dead letters programadas periódicamente
Futuro Extensión del patrón a la etapa de transformación (extract_* functions)
Futuro Extensión a la etapa de MERGE, con captura de registros que fallan en la actualización de Delta

8. Conclusión

La implementación del Dead Letter Pattern en el pipeline PLACSP resuelve un problema fundamental de resiliencia: la capacidad de procesar datos de una fuente externa heterogénea e impredecible sin que un único registro corrupto bloquee el pipeline completo. La solución ha evolucionado desde una lectura única a nivel de <feed> hacia una lectura dual a nivel de <entry>, con tres cambios fundamentales que trabajan conjuntamente:

  1. El cambio de granularidad (de rowTag="feed" a rowTag="entry" y rowTag="at:deleted-entry") reduce la unidad de fallo al registro individual y elimina la necesidad del explode() en las funciones de extracción.

  2. La corrección de los mismatches estructurales en bidding_schema (tres campos cambiados de ArrayType a StructType) elimina el comportamiento de eliminación silenciosa de entries que columnNameOfCorruptRecord exhibe ante discrepancias de tipo estructural.

  3. El requisito de nullable=True en todos los campos de alto nivel del esquema permite que el encoder de Spark serialice sin errores las filas corruptas (donde todos los campos son null) antes de que el filtro _corrupt_record.isNotNull() las desvíe al dead letter. Sin esta propiedad, el encoder lanza assertnotnull al cruzar fronteras de stage, abortando el pipeline antes de que el filtro pueda actuar.

El resultado es un pipeline que cumple las tres propiedades fundamentales del patrón: no bloqueo, trazabilidad y replayabilidad, estableciendo las bases para una operación autónoma robusta bajo orquestación con Apache Airflow.