Declarative Data Pipeline in Spark | Classes and Decorators

Declarative Approach to Building Data Pipelines: Classes and Decorators in Spark

In data engineering projects, business logic often becomes fragmented due to the lack of a unified architecture. The proposed method leverages Python metaprogramming to create declarative pipelines, where data transformations are defined using classes and decorators. This approach simplifies maintenance, boosts code reusability, and makes business processes explicit.

Core Principles of Declarative Pipelines

The key idea is to treat the business process (Flow) as the primary entity, while viewing tables as auxiliary objects. This prevents logic dilution as the system scales. A Flow consists of a sequence of steps, each of which:

• Takes input tables
• Performs a transformation
• Returns the result

Importantly, tables are implemented as classes rather than primary objects. For example, behind the MyTable class might be a Spark DataFrame. This level of abstraction lets you change the underlying implementation without rewriting business logic. Steps must remain atomic with clear inputs and outputs to ensure predictable execution.

Implementation Using Classes and Decorators

The base Flow class houses the framework logic. Specific pipelines are created via inheritance:

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}")

Decorators don't execute code; they accumulate metadata:

• @Flow.step(order=1) — defines execution order
• @Flow.input() — specifies input tables
• @Flow.output() — sets the output

When a subclass is created, the __init_subclass__ method kicks in, automatically collecting steps, sorting them by order, and building the executable sequence. This is done by inspecting class attributes and extracting metadata from decorated methods.

State Management: Context and Tables

Data is passed between steps using a Context object, implemented as a Pydantic model:

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

Advantages of this approach:

• Minimizes method parameters
• Provides centralized state management
• Offers flexibility for extensions

Tables are defined using SQLAlchemy's declarative style. A converter transforms ORM models into Spark schemas:

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)

This enables a unified data model across system layers. If needed, you can override the schema directly in the get_schema() method.

Running the Pipeline and SQL Processing

Pipeline execution happens via the run() method, which calls steps sequentially in the specified order. SQL is recommended as the primary tool for transformations:

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})

The @Flow.sql decorator prepares a parameterized query. Before execution, SQL automatically creates temporary views for input tables using the create_temp_views() method. This allows seamless use of tables in queries without manual context handling.

Key Points

• Declarative style makes business logic explicit and simplifies debugging
• Context as a single access point for state reduces errors
• Prioritizing SQL over DataFrame API speeds up development and improves readability
• Metaprogramming via __init_subclass__ automates pipeline assembly

— Editorial Team