Authoring des tâches Go SDK

Le Airflow Go SDK implémente le modèle language-SDK pour Go : votre DAG reste en Python, et chaque tâche est une fonction Go compilée enregistrée dans un bundle (un seul exécutable natif). Cette compétence couvre l'API native spécifique à Go. Le modèle partagé (le pattern Python @task.stub, la correspondance des IDs, le contrat XCom-as-JSON) se trouve dans authoring-language-sdk-tasks ; lisez-le d'abord si vous êtes nouveau aux language SDKs.

Expérimental. Le Go SDK est en développement actif et n'est pas prêt pour la production. Module path github.com/apache/airflow/go-sdk (Go 1.24+). Les APIs peuvent changer.

Compétences associées : authoring-language-sdk-tasks (stub Python partagé + concepts), deploying-go-sdk-bundles (construire, empaqueter et déployer le bundle), configuring-airflow-language-sdks (router la queue vers le coordinateur Go).

Récapitulatif : le côté Python

Une tâche Go est associée à un stub Python qui ne contient aucune logique ; il déclare la tâche, sa queue et le graphe de dépendances. Les IDs doivent correspondre exactement à l'enregistrement Go, et queue= route la tâche vers le runtime Go. Les règles complètes se trouvent dans authoring-language-sdk-tasks ; la forme minimale :

from airflow.sdk import dag, task


@task.stub(queue="golang")
def extract(): ...


@task.stub(queue="golang")
def transform(): ...


@dag()
def simple_dag():
    extract() >> transform()


simple_dag()

La valeur de queue ("golang" ici) est un label arbitraire qui doit correspondre à la queue routée vers le coordinateur Go (queue_to_coordinator). Voir configuring-airflow-language-sdks.

Le point d'entrée du bundle

Un bundle implémente bundlev1.BundleProvider : rapportez sa version et enregistrez vos DAGs et tâches. main est une ligne ; bundlev1server.Serve connecte le bundle au runtime Airflow pour vous.

package main

import (
    "log"

    v1 "github.com/apache/airflow/go-sdk/bundle/bundlev1"
    "github.com/apache/airflow/go-sdk/bundle/bundlev1/bundlev1server"
)

type myBundle struct{}

var _ v1.BundleProvider = (*myBundle)(nil)

func (m *myBundle) GetBundleVersion() v1.BundleInfo {
    return v1.BundleInfo{Name: bundleName, Version: &bundleVersion}
}

func (m *myBundle) RegisterDags(dagbag v1.Registry) error {
    simpleDag := dagbag.AddDag("simple_dag")      // dag_id must match the Python @dag name
    simpleDag.AddTask(extract)                    // task_id is the function name; must match the stub
    simpleDag.AddTaskWithName("transform", transform) // or set the task_id explicitly
    return nil
}

func main() {
    if err := bundlev1server.Serve(&myBundle{}); err != nil {
        log.Fatal(err)
    }
}

AddTask(fn) dérive le task_id du nom de la fonction Go ; utilisez AddTaskWithName("<task_id>", fn) quand ce nom ne peut pas correspondre au stub Python (une fonction non exportée, renommée ou réutilisée). RegisterDags est l'unique source de vérité pour l'identité des tâches : le manifest du bundle (utilisé par l'empaqueteur et par le coordinateur) est généré en l'exécutant, jamais écrit à la main.

Fonctions de tâche : injection de dépendances par type de paramètre

Une tâche est une fonction Go ordinaire. Le runtime inspecte sa signature et injecte des arguments par type ; déclarez seulement ce dont vous avez besoin.

La signature de retour optionnelle est (result, error) : un result non-nil est pushé comme le XCom return_value de la tâche ; un error non-nil échoue la tâche (ce qui déclenche la politique de retry du stub). Retourner seulement error, ou rien, est aussi valide.

func extract(ctx sdk.TIRunContext, client sdk.Client, log *slog.Logger) (any, error) {
    conn, err := client.GetConnection(ctx, "test_http")
    if err != nil {
        return nil, err
    }
    log.Info("connected", "host", conn.Host)
    return map[string]any{"go_version": runtime.Version()}, nil
}

func transform(ctx sdk.TIRunContext, client sdk.VariableClient) error {
    val, err := client.GetVariable(ctx, "my_variable")
    if err != nil {
        return err // VariableNotFound (a sentinel error) if absent
    }
    _ = val
    return nil
}

La surface de sdk.Client

Connection expose ID, Type, Host, Port (int), Login *string, Password *string (nil quand non défini, distinct d'une chaîne vide), Path (schéma), Extra map[string]any, plus GetURI(). Les cas non trouvés retournent les sentinelles sdk.VariableNotFound, sdk.ConnectionNotFound, sdk.XComNotFound.

Pour lire le résultat d'une tâche amont, appelez GetXCom explicitement, en prenant les dag_id/run_id/task_id dont vous avez besoin du contexte runtime (ci-dessous).

Contexte runtime

Déclarez un paramètre sdk.TIRunContext pour lire les métadonnées sur l'instance de tâche et sa DAG run. C'est une interface qui intègre context.Context, donc elle est utilisable partout où un context.Context est attendu.

func extract(ctx sdk.TIRunContext, log *slog.Logger) error {
    ti, dagRun := ctx.TaskInstance(), ctx.DagRun()
    log.Info("running",
        "task_id", ti.TaskID,
        "run_id", dagRun.RunID,
        "logical_date", dagRun.LogicalDate)
    return nil
}

• TaskInstance() : DagID, RunID, TaskID, MapIndex *int (nil quand non mappé), TryNumber.
• DagRun() : DagID, RunID, et les timestamps *time.Time LogicalDate, DataIntervalStart, DataIntervalEnd (nil quand non envoyés).

Les accesseurs sont remplis à partir des détails de démarrage de la tâche avant que le corps ne s'exécute. Parce que TIRunContext intègre context.Context, passez-le directement aux appels client et aux vérifications d'annulation (ctx.Done()) ; déclarez-le comme votre paramètre context par défaut. Dans les tests, construisez l'argument avec sdk.NewTIRunContext(ctx, ti, dagRun) (il panique sur un ctx nil).

Pièges spécifiques à Go

• Les IDs doivent correspondre au stub Python (dag_id de AddDag, task_id du nom de fonction enregistré), et la queue= du stub doit router vers le coordinateur Go, sinon la tâche n'est jamais livrée.
• RegisterDags est autoritaire. N'écrivez pas le manifest à la main ; l'empaqueteur le génère en exécutant RegisterDags.
• Demandez l'interface client la plus restreinte dont vous avez besoin (sdk.VariableClient plutôt que sdk.Client) pour une intention plus claire et des simulations plus faciles.
• Un retour error non-nil échoue la tâche et applique les retries du stub ; une panique récupérée est aussi un échec.
• Voir authoring-language-sdk-tasks pour les pièges indépendants du langage (un processus par instance de tâche, définissez queue et retries sur le stub).

Compétences associées

• authoring-language-sdk-tasks : Pattern et concepts partagés de stub Python (lisez d'abord).
• deploying-go-sdk-bundles : Construire et empaqueter le bundle avec go tool airflow-go-pack, puis le déployer pour le coordinateur.
• configuring-airflow-language-sdks : Router la queue vers le coordinateur Go (ExecutableCoordinator).
• authoring-dags : Authoring général des DAGs Airflow.