Deklaratywny Data Pipeline w Spark | Klasy i dekoratory

# Deklaratywny sposób budowania potoków danych: klasy i dekoratory w Spark

W projektach data engineering często dochodzi do fragmentacji logiki biznesowej z powodu braku jednolitej architektury. Proponowana metoda wykorzystuje metaprogramowanie Pythona do tworzenia deklaratywnych potoków danych, w których transformacje danych opisuje się za pomocą klas i dekoratorów. Takie podejście upraszcza utrzymanie kodu, zwiększa jego reużywalność i czyni procesy biznesowe jawne.

Podstawowe zasady deklaratywnego potoku danych

Kluczowy pomysł polega na wyodrębnieniu procesu biznesowego (Flow) jako głównej encji, a traktowaniu tabel jako obiektów pomocniczych. Pozwala to uniknąć rozmycia logiki przy skalowaniu. Flow to sekwencja kroków (steps), z których każdy:

• Przyjmuje tabele wejściowe
• Wykonuje transformację
• Zwraca wynik

Ważne jest, że tabele są implementowane jako klasy, a nie jako obiekty pierwotne. Na przykład za klasą MyTable może stać Spark DataFrame. Taki poziom abstrakcji umożliwia zmianę implementacji bez przepisywania logiki biznesowej. Kroki powinny być atomowe i mieć jasno określone wejścia/wyjścia, co zapewnia przewidywalność wykonania.

Implementacja za pomocą klas i dekoratorów

Podstawowa klasa Flow zawiera logikę frameworka. Konkretne potoki tworzy się przez dziedziczenie:

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

Dekoratory nie wykonują kodu, lecz gromadzą metainformacje:

• @Flow.step(order=1) — określa kolejność wykonania
• @Flow.input() — wskazuje tabele wejściowe
• @Flow.output() — definiuje wynik

Podczas tworzenia podklasy uruchamia się metoda __init_subclass__, która automatycznie zbiera kroki, sortuje je według kolejności i tworzy wykonywalną sekwencję. Osiąga się to poprzez analizę atrybutów klasy i wyodrębnienie metadanych z udekorowanych metod.

Zarządzanie stanem: kontekst i tabele

Do przekazywania danych między krokami służy obiekt Context, zaimplementowany za pomocą modelu Pydantic:

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

Zalety takiego podejścia:

• Minimalizacja parametrów w metodach
• Zcentralizowane zarządzanie stanem
• Elastyczność rozszerzeń

Tabele opisuje się w stylu deklaratywnym SQLAlchemy. Konwerter przekształca modele ORM w schematy 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)

Pozwala to używać jednolitego modelu danych w różnych warstwach systemu. W razie potrzeby schemat można nadpisać bezpośrednio w metodzie get_schema().

Uruchamianie potoku i obsługa SQL

Wykonanie potoku odbywa się za pomocą metody run(), która kolejno wywołuje kroki w określonej kolejności. Do transformacji zaleca się używanie SQL jako głównego narzędzia:

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

Dekorator @Flow.sql przygotowuje zapytanie parametryzowane. Przed wykonaniem SQL automatycznie tworzy tymczasowe widoki dla tabel wejściowych za pomocą metody create_temp_views(). Zapewnia to przezroczyste używanie tabel w zapytaniach bez ręcznego zarządzania kontekstem.

Co jest ważne

• Deklaratywność czyni logikę biznesową jawną i ułatwia debugowanie
• Kontekst jako jednolity punkt dostępu do stanu zmniejsza liczbę błędów
• Priorytet SQL nad API DataFrame przyspiesza rozwój i poprawia czytelność
• Metaprogramowanie za pomocą __init_subclass__ automatyzuje składanie potoku

— Editorial Team