Tuberías y filtros y arquitecturas de pipeline

Definición y origen

En el estilo de tuberías y filtros, el procesamiento es una cadena de pasos independientes. Cada filtro lee datos de su entrada, hace un solo trabajo con ellos y escribe el resultado en su salida. Cada tubería lleva los datos en un solo sentido de un filtro al siguiente, normalmente con un búfer entre ambos. La descripción de Garlan y Shaw (1994) enuncia las restricciones que dan al estilo sus propiedades: los filtros deben ser independientes, no deben compartir estado con otros filtros y no deben conocer la identidad de los filtros situados a ambos extremos de sus tuberías. Un filtro solo conoce el formato de lo que lee y de lo que escribe.

El hogar más conocido del estilo es Unix. En un memorándum fechado el 11 de octubre de 1964, Doug McIlroy escribió que debería haber formas de acoplar programas como una manguera de jardín, enroscando otro tramo cuando hay que tratar los datos de otra manera. Ken Thompson implementó la llamada al sistema pipe y las tuberías en el shell en la Version 3 de Unix, en 1973, y la notación | llegó en la Version 4. La filosofía Unix que surgió de ahí, escribir programas que hagan una sola cosa bien y trabajen juntos a través de una interfaz universal de texto, es el estilo de tuberías y filtros aplicado al conjunto de herramientas de todo un sistema operativo. POSA (1996) catalogó Pipes and Filters como patrón arquitectónico, y Enterprise Integration Patterns (2003), de Hohpe y Woolf, lo aplicó a la mensajería entre sistemas.

Richards y Ford llaman a este mismo estilo arquitectura de pipeline y distinguen cuatro tipos de filtro. Un productor (o fuente, source) inicia el flujo, leyendo de un archivo, una base de datos o una cola. Un transformador cambia cada registro: lo analiza, lo convierte, lo enriquece. Un tester comprueba una condición y deja pasar el registro o lo descarta, a veces enviándolo a una de varias salidas. Un consumidor (o sumidero, sink) termina el flujo, escribiendo en una base de datos, un archivo o una pantalla.

Cómo funciona

Las restricciones producen directamente las ventajas. Como un filtro no comparte estado y no conoce a sus vecinos, se puede escribir y probar por separado, con una lista de registros de entrada y una lista esperada de salidas; se puede reutilizar en otros pipelines; y se pueden añadir, quitar o reordenar filtros cambiando solo la forma en que se monta el pipeline. Como los registros fluyen de forma incremental, un filtro puede empezar a trabajar con el primer registro antes de que el filtro anterior haya terminado el último, así que las etapas se ejecutan de forma concurrente y todo el pipeline solo necesita memoria para unos pocos registros a la vez, no para todo el conjunto de datos.

La tubería oculta cómo se mueven los datos. En un shell de Unix es un búfer del sistema operativo entre dos procesos; en Linux admite 64 KiB por defecto, y un escritor que lo llena simplemente se bloquea hasta que el lector se pone al día. En un programa Python puede ser un generador, de modo que cada filtro pide el siguiente registro al anterior. En una plataforma de datos puede ser una cola de mensajes o un topic de un log como Apache Kafka, lo que permite que los filtros se ejecuten en máquinas distintas y sobrevivan a los reinicios. El código de los filtros apenas cambia entre estas opciones; las propiedades operativas cambian muchísimo.

Hay variaciones de la forma estricta. Un pipeline puede abrirse en abanico (fan-out: un tester envía registros a dos consumidores distintos) o converger (fan-in: dos productores alimentan a un transformador). Puede ser lineal, lo más fácil de razonar, o un grafo acíclico dirigido de pasos, como en herramientas de flujos de trabajo como Apache Airflow. En sentido estricto, un filtro con estado interno, como uno que cuenta u ordena, rompe la propiedad incremental, porque debe ver toda la entrada antes de producir salida; sort en un pipeline de Unix es un filtro de ese tipo. Estos filtros están permitidos, pero se convierten en los puntos donde hay que pensar en la memoria, la latencia y los reinicios.

Ejemplos: dónde aparecen los pipelines

ETL y pipelines de datos. Los trabajos de extracción, transformación y carga (ETL) mueven datos de los sistemas operacionales a un almacén de datos: extraer los pedidos de ayer, depurarlos y convertirlos, unirlos con los datos de los restaurantes y cargarlos en tablas de informes. Cada paso es un filtro, y las plataformas de datos modernas (Apache Beam, Spark, modelos de dbt, servicios de flujo de datos en la nube) son, en gran medida, herramientas para construir y ejecutar esos pipelines a escala.

Compiladores. Un compilador clásico es un pipeline de fases: el análisis léxico convierte caracteres en tokens, el análisis sintáctico convierte tokens en un árbol sintáctico, el análisis semántico comprueba tipos y nombres, la optimización reescribe una representación intermedia y la generación de código emite código máquina. Cada fase consume una representación y produce la siguiente, así que las fases se pueden desarrollar y sustituir por separado; el proyecto LLVM, por ejemplo, permite que muchos frontends de lenguajes compartan un optimizador y un conjunto de backends. Los compiladores reales relajan el estilo estricto, porque las fases comparten una tabla de símbolos.

Procesamiento de medios. La transcodificación de video y el procesamiento de audio encadenan decodificadores, escaladores, filtros y codificadores. GStreamer construye aplicaciones multimedia explícitamente como pipelines de elementos enlazados mediante pads, y los grafos de filtros de FFmpeg funcionan de la misma manera.

Procesamiento de logs y eventos. Los enviadores de logs como Logstash y Fluent Bit se configuran como etapas de entrada, filtro y salida: leen líneas de log, las descomponen en campos, descartan el ruido de las comprobaciones de estado (health checks), enmascaran los datos personales y las envían a un índice de búsqueda. Los datos de observabilidad de la plataforma de reparto de comida fluyen de esta manera.

Pipelines de características y de inferencia de aprendizaje automático. La preparación de los datos de entrenamiento, y el preprocesamiento y el posprocesamiento alrededor de un modelo, se construyen habitualmente como pipelines, a los que vuelve el Capítulo 12.

# Top ten URLs that returned HTTP 500 in today's access log
grep '" 500 ' access.log \
  | awk '{print $7}' \
  | sort \
  | uniq -c \
  | sort -rn \
  | head -10

import json

def read_lines(path):                 # producer
    with open(path) as f:
        for line in f:
            yield line

def parse(lines):                     # transformer
    for line in lines:
        yield json.loads(line)

def only_failed(events):              # tester
    for e in events:
        if e["status"] == "payment_failed":
            yield e

def count_by_provider(events):        # consumer
    counts = {}
    for e in events:
        counts[e["provider"]] = counts.get(e["provider"], 0) + 1
    return counts

events = only_failed(parse(read_lines("orders.jsonl")))
print(count_by_provider(events))

Por lotes y en streaming

La misma estructura funciona en dos modos. El procesamiento por lotes trata una entrada acotada, como todos los pedidos de ayer, se ejecuta hasta completarse y se detiene. Es fácil de razonar: si una ejecución falla, corriges el problema y la vuelves a ejecutar, siempre que cada ejecución sustituya su salida en lugar de añadirse a ella. Los pagos nocturnos a los restaurantes, las facturas mensuales y las cargas del almacén de datos son pipelines por lotes típicos. Su punto débil es el retraso: los resultados tienen la antigüedad de la última ejecución.

El procesamiento de flujos (stream processing) trata una entrada no acotada y procesa cada registro poco después de que llegue: posiciones de repartidores, eventos de pago, clics. Los resultados están frescos, pero aparecen nuevas preguntas. Un flujo nunca termina, así que los agregados deben calcularse sobre ventanas (pedidos por restaurante cada cinco minutos); los registros pueden llegar tarde o desordenados; y el pipeline debe funcionar de forma continua, recuperarse de las caídas sin perder registros ni contarlos dos veces, y aguantar cuando la tasa de entrada se dispara. Frameworks como Kafka Streams, Apache Flink y Apache Beam ofrecen ventanas, checkpoints y almacenes de estado para estos problemas. El modelo de Beam trata el procesamiento por lotes como un caso especial del streaming, con una entrada acotada, así que una misma definición de pipeline puede ejecutarse de las dos formas.

Muchas organizaciones usan ambos: un flujo para paneles operativos frescos y aproximados, y un trabajo por lotes que recalcula las cifras exactas durante la noche. La Sección 7.8 analiza el lado de arquitectura de datos de esta elección.

Back pressure y gestión de errores

Cuando un productor es más rápido que un consumidor, algo tiene que ceder. Las opciones son usar un búfer (lo que solo aplaza el problema, a menos que el búfer esté acotado), descartar registros o tomar muestras (aceptable para algunas métricas, nunca para pagos), o aplicar back pressure (contrapresión): hacer que la etapa rápida baje a la velocidad de la lenta. Las tuberías de Unix lo hacen automáticamente, porque una tubería llena bloquea al escritor. Los diseños basados en extracción (pull), como los generadores de Python, lo hacen por construcción, porque no se produce nada hasta que la siguiente etapa lo pide. Los pipelines distribuidos necesitan que se diseñe expresamente: colas acotadas, consumidores que extraen de un log a su propio ritmo, o protocolos como la especificación Reactive Streams de la JVM, en la que un suscriptor indica cuántos registros está listo para recibir. La Sección 9.3 trata a fondo las colas y la back pressure.

Los errores necesitan una política para cada filtro, decidida de antemano. Un registro mal formado (un mensaje envenenado, poison message) no debe detener el pipeline: envíalo a una cola de mensajes fallidos (dead-letter queue) o a un archivo de errores con el motivo, cuéntalo y lanza una alerta si la tasa aumenta. Un fallo transitorio, como que un servicio de consulta agote su tiempo de espera, se reintenta con backoff. Una caída del pipeline necesita un punto de reinicio: los trabajos por lotes se vuelven a ejecutar desde el principio o desde un checkpoint, y los frameworks de streaming guardan en checkpoints su posición y su estado. Como los reinicios implican que algunos registros se procesen dos veces, los consumidores que escriben resultados deben ser idempotentes, por ejemplo haciendo un upsert sobre una clave natural en lugar de insertar a ciegas. Los errores son el punto más débil del estilo: los filtros no se conocen entre sí a propósito, así que un fallo descubierto en el paso cuatro no puede deshacer fácilmente lo que ya hizo el paso dos, y un pipeline que necesite ese tipo de reversión coordinada probablemente usa el estilo equivocado.

Cuándo usarlo, cuándo no y antipatrones

Usa tuberías y filtros cuando el problema sea por naturaleza un flujo de datos en un solo sentido a través de transformaciones distintas: ingesta, conversión, enriquecimiento, validación, agregación, exportación. Es la opción por defecto en ingeniería de datos, en el procesamiento de archivos y medios, y en cualquier trabajo en el que los pasos cambien más a menudo que el flujo general. También aparece dentro de otros estilos: un trabajo nocturno de informes dentro de un monolito modular, o un servicio de procesamiento de flujos dentro de un sistema dirigido por eventos.

Evítalo como estructura principal de aplicaciones interactivas, en las que los usuarios esperan respuestas inmediatas y conversacionales y el trabajo no es un flujo; para lógica en la que muchos pasos necesitan compartir y actualizar el mismo estado; y para procesos que necesitan una reversión coordinada entre pasos, que requieren una transacción o una saga.

Antipatrones habituales: el filtro dios, un paso que analiza, enriquece, valida y escribe, con lo que se pierden las ventajas del estilo; el estado compartido oculto, filtros que se comunican a través de una tabla de base de datos compartida o una variable global en lugar de la tubería, lo que reintroduce el acoplamiento que el estilo elimina; la deriva de formato, un productor que cambia sin avisar su formato de salida y rompe todos los filtros posteriores, algo que se evita con esquemas explícitos y comprobaciones de esquema (consulta la Sección 7.7); los búferes no acotados, que convierten a un consumidor lento en una caída por falta de memoria; y los sumideros no idempotentes, que cuentan dos veces después de cada reintento.

Hacer cumplir el estilo en el código

Las restricciones del estilo se pueden hacer explícitas. Da a cada filtro la misma firma sencilla (entra un iterable de registros, sale un iterable), para que la composición sea mecánica. Define el formato de los registros en cada tubería como un esquema (una dataclass, un JSON Schema, una definición de Avro o de Protocol Buffers) y valídalo en el límite. Mantén los filtros libres de estado mutable compartido: una prueba de arquitectura puede prohibir que los módulos de filtros importen la capa de base de datos o se importen entre sí, de modo que la única forma de que los datos pasen entre ellos sea la tubería. Prueba cada filtro con pequeños datos de prueba (fixtures) de entrada y salida, y prueba el pipeline montado de extremo a extremo con una muestra representativa, que incluya registros mal formados, para demostrar que la política de errores funciona.

En los pipelines distribuidos, monitoriza el throughput por etapa, la profundidad de las colas y el retraso de los consumidores (consumer lag): una cola que crece delante de una etapa es la señal visible de un cuello de botella o de un filtro que falla, y es la primera métrica sobre la que alertar.