authoring-language-sdk-tasks

Authoring Language SDK Tasks (Shared Foundation)

Airflow language SDKs let you implement task logic in a language other than Python while the DAG and its scheduling stay in Python. This skill describes the parts that are identical across every language SDK. Each language has its own companion skill for the native API, build tooling, and runtime — see Per-language skills.

Experimental. The language SDKs are in preview. APIs and artifact coordinates may change.

---

The model

A DAG is authored in Python as usual. Tasks that should run in another language are declared as stubs routed to a dedicated queue. At runtime, Airflow hands a stub task to a coordinator that launches a short-lived native subprocess for that one task instance, runs your compiled/native code, and shuts the subprocess down.

Consequences that hold for every language SDK:

• One subprocess per task instance — there is no shared in-process state between task instances. Pass data via XCom or an external store.
• The DAG, schedule, retries, and queue routing live in Python. The native side only implements task logic.
• Data crossing the boundary is JSON. See The XCom-as-JSON contract.

---

The two-sided model

Every task has two halves that must agree:

1. A Python stub in a normal DAG file — no logic; it declares the task, its queue, the dependency graph, and retry policy.
2. A native implementation (Java, Go, etc.) whose IDs match the Python side and where the work happens.

Python side (scheduling)

The example below uses the Go SDK to be concrete, but the Python side is identical for every language SDK. The queue name ("golang" here) is an arbitrary label you choose — it just has to match a key in queue_to_coordinator (see configuring-airflow-language-sdks). Pick whatever name fits the SDK you're routing to.

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