Pipeline de Données Déclarative dans Spark | Classes et Décorateurs

# Approche déclarative pour la construction de pipelines de données : classes et décorateurs dans Spark

Dans les projets d'ingénierie des données, la logique métier se fragmente souvent en raison de l'absence d'une architecture unifiée. La méthode proposée exploite la métaprogrammation Python pour créer des pipelines déclaratifs, où les transformations de données sont définies à l'aide de classes et de décorateurs. Cette approche simplifie la maintenance, améliore la réutilisabilité du code et rend les processus métier explicites.

Principes fondamentaux des pipelines déclaratifs

L'idée clé est de considérer le processus métier (Flow) comme l'entité principale, tandis que les tables sont vues comme des objets auxiliaires. Cela évite la dilution de la logique à mesure que le système s'étend. Un Flow consiste en une séquence d'étapes, chacune d'elles :

• Prend des tables en entrée
• Effectue une transformation
• Retourne le résultat

Important : les tables sont implémentées comme des classes plutôt que comme des objets primaires. Par exemple, derrière la classe MyTable se trouve peut-être un Spark DataFrame. Ce niveau d'abstraction permet de changer l'implémentation sous-jacente sans réécrire la logique métier. Les étapes doivent rester atomiques avec des entrées et sorties claires pour assurer une exécution prévisible.

Implémentation avec classes et décorateurs

La classe de base Flow contient la logique du framework. Les pipelines spécifiques sont créés par héritage :

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

Les décorateurs n'exécutent pas de code ; ils accumulent des métadonnées :

• @Flow.step(order=1) — définit l'ordre d'exécution
• @Flow.input() — spécifie les tables en entrée
• @Flow.output() — définit la sortie

Lors de la création d'une sous-classe, la méthode __init_subclass__ entre en action, collectant automatiquement les étapes, les triant par ordre et construisant la séquence exécutable. Cela se fait en inspectant les attributs de la classe et en extrayant les métadonnées des méthodes décorées.

Gestion d'état : Context et tables

Les données sont transmises entre les étapes via un objet Context, implémenté comme un modèle Pydantic :

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

Avantages de cette approche :

• Minimise les paramètres des méthodes
• Fournit une gestion centralisée de l'état
• Offre une flexibilité pour les extensions

Les tables sont définies en utilisant le style déclaratif de SQLAlchemy. Un convertisseur transforme les modèles ORM en schémas 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)

Cela permet un modèle de données unifié à travers les couches du système. Si nécessaire, vous pouvez surcharger le schéma directement dans la méthode get_schema().

Exécution du pipeline et traitement SQL

L'exécution du pipeline se fait via la méthode run(), qui appelle les étapes de manière séquentielle dans l'ordre spécifié. SQL est recommandé comme outil principal pour les 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})

Le décorateur @Flow.sql prépare une requête paramétrée. Avant l'exécution, SQL crée automatiquement des vues temporaires pour les tables en entrée en utilisant la méthode create_temp_views(). Cela permet une utilisation transparente des tables dans les requêtes sans gestion manuelle du contexte.

Points clés

• Le style déclaratif rend la logique métier explicite et simplifie le débogage
• Context comme point d'accès unique à l'état réduit les erreurs
• Prioriser SQL plutôt que l'API DataFrame accélère le développement et améliore la lisibilité
• La métaprogrammation via __init_subclass__ automatise l'assemblage du pipeline

— Editorial Team