Pipeline de Datos Declarativo en Spark | Clases y Decoradores

# Enfoque declarativo para construir pipelines de datos: Clases y decoradores en Spark

En los proyectos de ingeniería de datos, la lógica de negocio a menudo se fragmenta debido a la falta de una arquitectura unificada. El método propuesto aprovecha la metaprogramación de Python para crear pipelines declarativos, en los que las transformaciones de datos se definen mediante clases y decoradores. Este enfoque simplifica el mantenimiento, aumenta la reutilización del código y hace explícitos los procesos de negocio.

Principios fundamentales de los pipelines declarativos

La idea clave es tratar el proceso de negocio (Flow) como la entidad principal, mientras que las tablas se consideran objetos auxiliares. Esto evita la dilución de la lógica a medida que el sistema escala. Un Flow consiste en una secuencia de pasos, cada uno de los cuales:

• Toma tablas de entrada
• Realiza una transformación
• Devuelve el resultado

Es importante que las tablas se implementen como clases en lugar de objetos primarios. Por ejemplo, detrás de la clase MyTable podría haber un Spark DataFrame. Este nivel de abstracción permite cambiar la implementación subyacente sin reescribir la lógica de negocio. Los pasos deben permanecer atómicos, con entradas y salidas claras, para garantizar una ejecución predecible.

Implementación con clases y decoradores

La clase base Flow alberga la lógica del framework. Los pipelines específicos se crean mediante herencia:

class MyFlow(Flow):
    @classmethod
    @Flow.step(order=1)
    @Flow.input([MyTable])
    @Flow.output(MyTable2)
    def step_one(cls, context: Context) -> DataFrame:
        print(f" Step 1: sozdayom/update MyTable,{context.id}")

Los decoradores no ejecutan código; acumulan metadatos:

• @Flow.step(order=1) — define el orden de ejecución
• @Flow.input() — especifica las tablas de entrada
• @Flow.output() — establece la salida

Al crear una subclase, el método __init_subclass__ entra en acción, recolectando automáticamente los pasos, ordenándolos por secuencia y construyendo la secuencia ejecutable. Esto se logra inspeccionando los atributos de la clase y extrayendo metadatos de los métodos decorados.

Gestión de estado: Context y tablas

Los datos se pasan entre pasos mediante un objeto Context, implementado como un modelo de Pydantic:

class Context(BaseModel):
    config: Dict[Type[Config], Config] = {}
    data: Dict[Union[str, Type[Table]], DataFrame] = {}
    diff: Dict[Any, Any] = {}

Ventajas de este enfoque:

• Minimiza los parámetros de los métodos
• Proporciona una gestión centralizada del estado
• Ofrece flexibilidad para extensiones

Las tablas se definen usando el estilo declarativo de SQLAlchemy. Un convertidor transforma los modelos ORM en esquemas de Spark:

class Table(Base):
    __abstract__ = True

    @classmethod
    def get_schema(cls) -> T.StructType:
        fields = []
        for column in cls.__table__.columns:
            spark_type = _convert_type(column.type)
            fields.append(
                T.StructField(column.name, spark_type, column.nullable)
            )
        return T.StructType(fields)

Esto permite un modelo de datos unificado en todas las capas del sistema. Si es necesario, puedes sobrescribir el esquema directamente en el método get_schema().

Ejecución del pipeline y procesamiento SQL

La ejecución del pipeline se realiza mediante el método run(), que invoca los pasos de forma secuencial en el orden especificado. Se recomienda SQL como herramienta principal para las transformaciones:

class MyFlow(Flow):
    @classmethod
    @Flow.step(order=3)
    @Flow.input([MyTable])
    @Flow.output(AnotherTable)
    @Flow.sql("step_three.sql")
    def step_three(cls, context: Context):
        df = cls.execute_sql(context, "step_three.sql", vars={"id": 1})

El decorador @Flow.sql prepara una consulta parametrizada. Antes de la ejecución, SQL crea automáticamente vistas temporales para las tablas de entrada mediante el método create_temp_views(). Esto permite usar las tablas de manera fluida en las consultas sin manejo manual del contexto.

Puntos clave

• El estilo declarativo hace explícita la lógica de negocio y simplifica la depuración
• Context como punto de acceso único al estado reduce errores
• Priorizar SQL sobre la API de DataFrame acelera el desarrollo y mejora la legibilidad
• La metaprogramación mediante __init_subclass__ automatiza el ensamblaje del pipeline

— Editorial Team