Deklarative Datenpipeline in Spark | Klassen und Dekoratoren

Deklarativer Ansatz zum Aufbau von Datenpipelines: Klassen und Dekoratoren in Spark

In Datenengineering-Projekten wird die Geschäftslogik oft fragmentiert, da eine einheitliche Architektur fehlt. Die vorgeschlagene Methode nutzt Python-Metaprogrammierung, um deklarative Pipelines zu erstellen, in denen DatenTransformationen mit Klassen und Dekoratoren definiert werden. Dieser Ansatz vereinfacht die Wartung, erhöht die Wiederverwendbarkeit des Codes und macht die Geschäftsprozesse explizit.

Kernprinzipien deklarativer Pipelines

Die Schlüsselidee besteht darin, den Geschäftsprozess (Flow) als primäre Entität zu betrachten, während Tabellen als Hilfsobjekte gesehen werden. Dies verhindert eine Verwässerung der Logik beim Skalieren des Systems. Ein Flow besteht aus einer Abfolge von Schritten, von denen jeder:

• Eingabetabellen übernimmt
• eine Transformation ausführt
• das Ergebnis zurückgibt

Wichtig: Tabellen werden als Klassen implementiert, nicht als Primärobjekte. Beispielsweise könnte hinter der MyTable-Klasse ein Spark DataFrame stehen. Diese Abstraktionsebene ermöglicht es, die zugrunde liegende Implementierung zu ändern, ohne die Geschäftslogik umzuschreiben. Schritte müssen atomar bleiben mit klaren Eingaben und Ausgaben, um eine vorhersehbare Ausführung zu gewährleisten.

Implementierung mit Klassen und Dekoratoren

Die BasisFlow-Klasse enthält die Framework-Logik. Spezifische Pipelines entstehen durch Vererbung:

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

Dekoratoren führen keinen Code aus; sie sammeln Metadaten:

• @Flow.step(order=1) — definiert die Ausführungsreihenfolge
• @Flow.input() — spezifiziert Eingabetabellen
• @Flow.output() — legt die Ausgabe fest

Beim Erstellen einer Unterklasse greift die __init_subclass__-Methode, sammelt automatisch die Schritte, sortiert sie nach Reihenfolge und baut die ausführbare Sequenz auf. Dies geschieht durch Inspektion der Klassenattribute und Extraktion von Metadaten aus dekorierten Methoden.

Zustandsverwaltung: Context und Tabellen

Daten werden zwischen Schritten über ein Context-Objekt weitergereicht, implementiert als Pydantic-Modell:

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

Vorteile dieses Ansatzes:

• Minimiert Methodenparameter
• Bietet zentralisierte Zustandsverwaltung
• Ermöglicht Flexibilität für Erweiterungen

Tabellen werden im deklarativen Stil von SQLAlchemy definiert. Ein Konverter transformiert ORM-Modelle in Spark-Schemata:

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)

Dies ermöglicht ein einheitliches Datenmodell über alle Systemschichten hinweg. Falls nötig, können Sie das Schema direkt in der get_schema()-Methode überschreiben.

Ausführung der Pipeline und SQL-Verarbeitung

Die Pipeline-Ausführung erfolgt über die run()-Methode, die die Schritte sequentiell in der angegebenen Reihenfolge aufruft. SQL wird als primäres Werkzeug für Transformationen empfohlen:

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

Der @Flow.sql-Dekorator bereitet eine parametrisierte Abfrage vor. Vor der Ausführung erstellt SQL automatisch temporäre Views für Eingabetabellen unter Verwendung der create_temp_views()-Methode. Dies erlaubt die nahtlose Verwendung von Tabellen in Abfragen ohne manuelle Kontextbehandlung.

Wichtige Punkte

• Der deklarative Stil macht die Geschäftslogik explizit und vereinfacht das Debugging
• Context als zentraler Zugriffspunkt für den Zustand reduziert Fehler
• Priorisierung von SQL gegenüber der DataFrame-API beschleunigt die Entwicklung und verbessert die Lesbarkeit
• Metaprogrammierung über __init_subclass__ automatisiert den Pipeline-Aufbau

— Editorial Team