Deklarativní Data Pipeline ve Spark | Třídy a dekorátory

Deklarativní přístup k tvorbě data pipeline: třídy a dekorátory v Spark

V projektech data engineering často dochází k fragmentaci obchodní logiky kvůli absenci jednotné architektury. Navrhovaná metoda využívá metaprogramování Pythonu k vytvoření deklarativních pipelineů, kde transformace dat jsou popsány prostřednictvím tříd a dekorátorů. Tento přístup zjednodušuje údržbu, zvyšuje znovupoužitelnost kódu a činí obchodní procesy explicitními.

Základní principy deklarativního pipeline

Klíčový nápad spočívá v vyčlenění obchodního procesu (Flow) jako hlavní entity a považování tabulek za pomocné objekty. To zabraňuje rozmazání logiky při škálování. Flow představuje sekvenci kroků (steps), z nichž každý:

• Přijímá vstupní tabulky
• Provádí transformaci
• Vrátí výsledek

Důležité je, že tabulky jsou realizovány jako třídy, nikoli jako primární objekty. Například za třídou MyTable může stát Spark DataFrame. Tato úroveň abstrakce umožňuje měnit implementaci bez přepisování obchodní logiky. Kroky musí být atomární a mít jasné vstupy/výstupy pro zajištění předvídatelnosti provedení.

Realizace prostřednictvím tříd a dekorátorů

Základní třída Flow obsahuje logiku frameworku. Konkrétní pipeline se vytvářejí dědičností:

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: vytváříme/aktualizujeme MyTable,{context.id}")

Dekorátory neprovádějí kód, ale akumulují metainformace:

• @Flow.step(order=1) — určuje pořadí provedení
• @Flow.input() — specifikuje vstupní tabulky
• @Flow.output() — definuje výstup

Při vytváření podtřídy se aktivuje metoda __init_subclass__, která automaticky shromáždí kroky, seřadí je podle pořadí a vytvoří proveditelnou sekvenci. To se dosahuje analýzou atributů třídy a extrakcí metadat z dekorovaných metod.

Řízení stavu: kontext a tabulky

Pro předávání dat mezi kroky se používá objekt Context, realizovaný jako Pydantic model:

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

Výhody tohoto přístupu:

• Minimalizace parametrů v metodách
• Centralizované řízení stavu
• Flexibilita rozšiřování

Tabulky se popisují deklarativním stylem SQLAlchemy. Konvertor převádí ORM modely na Spark schémata:

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)

To umožňuje používat jednotný datový model pro různé vrstvy systému. Při potřebě lze schéma přepsat přímo v metodě get_schema().

Spuštění pipeline a zpracování SQL

Provedení pipeline probíhá metodou run(), která sekvenčně volá kroky v zadaném pořadí. Pro transformace se doporučuje používat SQL jako hlavní nástroj:

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

Dekorátor @Flow.sql připravuje parametrizovaný dotaz. Před provedením SQL automaticky vytváří dočasná zobrazení pro vstupní tabulky metodou create_temp_views(). To zajišťuje průhledné používání tabulek v dotazech bez ručního řízení kontextu.

Co je důležité

• Deklarativnost činí obchodní logiku explicitní a usnadňuje ladění
• Kontext jako jediný bod přístupu ke stavu snižuje počet chyb
• Priorita SQL před DataFrame API zrychluje vývoj a zlepšuje čitelnost
• Metaprogramování prostřednictvím __init_subclass__ automatizuje sestavení pipeline

— Editorial Team