Authoring Language SDK Tasks (Shared Foundation)

Les language SDKs Airflow vous permettent d'implémenter la logique des tâches dans un langage autre que Python tandis que le DAG et son ordonnancement restent en Python. Cette compétence décrit les parties identiques dans chaque language SDK. Chaque langage a sa propre compétence associée pour l'API native, les outils de build et le runtime — voir Per-language skills.

Experimental. Les language SDKs sont en préversion. Les APIs et les coordonnées d'artefacts peuvent changer.

The model

Un DAG est créé en Python comme d'habitude. Les tâches qui doivent s'exécuter dans un autre langage sont déclarées comme des stubs routés vers une queue dédiée. À l'exécution, Airflow transmet une tâche stub à un coordinator qui lance un court subprocess natif pour cette instance de tâche, exécute votre code compilé/natif et ferme le subprocess.

Les conséquences qui s'appliquent à chaque language SDK :

Un subprocess par instance de tâche — il n'y a pas d'état partagé en mémoire entre les instances de tâches. Transmettez les données via XCom ou un magasin externe.
Le DAG, l'ordonnancement, les retries et le routage des queues vivent en Python. Le côté natif n'implémente que la logique des tâches.
Les données franchissant la limite sont en JSON. Voir The XCom-as-JSON contract.

The two-sided model

Chaque tâche a deux moitiés qui doivent être cohérentes :

Un stub Python dans un fichier DAG normal — pas de logique ; il déclare la tâche, sa queue, le graphe de dépendances et la politique de retry.
Une implémentation native (Go, etc.) dont les IDs correspondent au côté Python et où le travail s'effectue.

Python side (scheduling)

L'exemple ci-dessous utilise le Go SDK pour être concret, mais le côté Python est identique pour chaque language SDK. Le nom de la queue ("golang" ici) est une étiquette arbitraire que vous choisissez — il doit simplement correspondre à une clé dans queue_to_coordinator (voir configuring-airflow-language-sdks). Choisissez le nom qui correspond au SDK vers lequel vous routez.

from datetime import timedelta
from airflow.sdk import dag, task


@dag
def sales_pipeline():
    @task.stub(queue="golang")          # queue selects the coordinator (see configuring-airflow-language-sdks)
    def extract(): ...

    @task.stub(queue="golang")
    def transform(extracted): ...        # arg only declares the dependency

    @task.stub(queue="golang", retries=1, retry_delay=timedelta(seconds=5))
    def load(transformed): ...

    @task()                              # an ordinary Python task can sit downstream
    def report(loaded):
        print(f"done: {loaded}")

    report(load(transform(extract())))


sales_pipeline()

Les règles qui s'appliquent indépendamment du langage :

Le nom de la fonction stub est l'ID de la tâche et le nom @dag (ou dag_id=) est l'ID du DAG. Le côté natif doit utiliser ces IDs exacts.
Un argument en amont sur un stub (par ex. transform(extracted)) existe uniquement pour déclarer la dépendance en Python. La valeur elle-même est récupérée du côté natif via XCom — la passer en Python ne la transmet pas au code natif.
La queue, les retries et autres arguments de tâche sont définis sur le stub, pas dans le code natif. Une tâche native qui échoue est rapportée à Airflow, qui applique ensuite la politique de retry du stub.
La valeur queue est ce qui route la tâche vers un coordinator ; la même chaîne doit apparaître dans queue_to_coordinator (voir configuring-airflow-language-sdks).

The XCom-as-JSON contract

Les valeurs XCom sont stockées en JSON dans la base de données de métadonnées d'Airflow, donc la limite entre Python et tout langage natif est JSON. Le côté Python/JSON est le même pour chaque SDK :

Type Python	JSON
`int`	number (integer)
`float`	number (decimal)
`str`	string
`bool`	boolean
`None`	null
`list`	array
`dict`	object

Chaque language SDK mappe ces types JSON sur ses propres types natifs. Le mapping des types natifs vit dans la compétence de ce langage. La règle de portabilité clé : une valeur envoyée par une tâche est lue par une autre en JSON, donc le côté consommateur doit s'attendre à un type compatible avec ce qui a été stocké.

What is language-specific (and lives elsewhere)

Cette compétence s'arrête délibérément aux concepts partagés. Les éléments suivants diffèrent selon le langage et sont documentés dans les compétences associées à chaque langage :

Native task API — comment vous déclarez les tâches, lisez les connexions/variables/XComs et envoyez les résultats (annotations, interfaces, enregistrement de fonctions, etc.).
Native type mapping — la colonne native du tableau JSON ci-dessus.
Build and packaging — comment l'artefact est compilé et bundlé.
Runtime prerequisite — ce qui doit être présent sur le worker (un runtime de langage pour certains SDKs ; rien pour les bundles auto-contenus du Go SDK).

Le câblage côté Airflow (quel coordinator exécute quelle queue) est partagé dans sa structure mais a des options par coordinateur ; il vit dans configuring-airflow-language-sdks.

Language-agnostic pitfalls

Les IDs doivent correspondre exactement entre le nom de la fonction stub Python et l'ID de tâche natif, et entre @dag/dag_id et l'ID du DAG natif. Les décalages se manifestent par des erreurs « no DAGs » ou XCom manquante.
Les deux côtés ont besoin de la référence en amont. Python déclare la dépendance en passant l'appel en amont ; le code natif récupère la valeur via XCom.
Définissez la queue et les retries sur le stub, jamais dans le code natif.
Les corps de stub doivent être vides. Une vérification AST l'impose — seul pass, ... ou une docstring sont autorisés dans le corps ; toute logique réelle est rejetée.
retry_policy est rejeté sur les stubs (@task.stub lève ValueError). Utilisez plutôt retries/retry_delay — un callable retry-policy exécute Python en mémoire et ne déclencherait jamais pour une tâche s'exécutant dans un subprocess natif.
Les assets, la deferral et quelques autres fonctionnalités Airflow ont un support limité ou inexistant dans les language SDKs aujourd'hui.

Per-language skills

authoring-go-sdk-tasks: Go native API — enregistrement des tâches, injection de dépendances par type de paramètre et accès client.
(Les futurs language SDKs ajoutent chacun leur propre compétence authoring-<lang>-sdk-tasks qui s'appuie sur celle-ci.)

Related Skills

configuring-airflow-language-sdks: Routez une queue vers un coordinator et définissez les options de runtime.
authoring-dags: Création générale de DAG Airflow (le côté Python vit ici aussi).
deploying-go-sdk-bundles: Construire, packager et livrer le bundle Go (les compétences de déploiement par langage suivent la même forme).

authoring-language-sdk-tasks