This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 40cc196f740 [v3-4-test] Go SDK: Rewrite the user docs around Dag
definition (#74698) (#75383)
40cc196f740 is described below
commit 40cc196f7404ea1381f43742e65d6bf5f11d0f44
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Oct 22 18:31:25 2026 +0800
[v3-4-test] Go SDK: Rewrite the user docs around Dag definition (#74698)
(#75383)
* Go SDK: Rewrite the user docs around Dag definition
Rewrite the Go SDK guide to cover declaring a Dag in Go first, then stub
TaskHandlers. Trim the README to a user-facing landing page, move the
contributor sections to DEVELOPMENT.md, and add a README for the examples.
* Drop the beta status from the Go SDK docs
* Explain the trigger rule a task after both If sides needs in the Go SDK
guide
* Apply suggestion from @pierrejeambrun
* Tighten the If wording, the LocalExecutor note and the duplicate config
block
---------
(cherry picked from commit b3db4d5201a25031fcb5b524936b471120d892ef)
Co-authored-by: Jason(Zhe-You) Liu
<[email protected]>
Co-authored-by: Pierre Jeambrun <[email protected]>
---
.../authoring-and-scheduling/language-sdks/go.rst | 997 ++++++++++++---------
go-sdk/DEVELOPMENT.md | 215 +++++
go-sdk/README.md | 507 ++---------
go-sdk/example/README.md | 129 +++
4 files changed, 953 insertions(+), 895 deletions(-)
diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst
b/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst
index fac17348007..8e87a094aa4 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst
@@ -20,270 +20,464 @@
Go SDK
======
-|experimental|
+The Go SDK lets you write Airflow Dags and tasks in Go. There are two ways to
use it:
-The Go SDK lets you implement Airflow task logic in Go, with native access to
the Airflow "model"
-(Variables, Connections, and XCom). The Dag and its scheduling remain in
Python; individual tasks delegate
-to a compiled Go *bundle* that is launched by
-:class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator` for each
task instance.
+* **Dag definition.** The schedule, the tasks, their options and the
dependencies between them are all written
+ in Go, with no Python file involved. See :ref:`go-sdk/dag-definition`.
+* **Python Dag with stub TaskHandler.** A Python Dag declares the tasks with
``@task.stub`` and wires them
+ together, and a Go ``TaskHandler`` implements each one. See
:ref:`go-sdk/stub-tasks`.
-Because Go is a compiled language, every task must be compiled ahead of time
and registered inside a single,
-self-contained native executable called a **bundle**. The bundle also embeds
the source file of each Dag built with ``airflow.Dag``
-plus the entrypoint file (the one with ``func main``), and a metadata
-manifest (the ``dag_id`` and ``task_id`` map) in a footer appended to the
executable, so the executable *is*
-the bundle: one runnable file to ship, with no separate manifest or archive.
The
-:ref:`airflow-go-pack <go-sdk/build>` tool builds and packs that bundle.
+Both use the same task API and the same build tool, and one bundle can serve
both.
+
+Go is compiled, so your Dags and tasks are built into a single self-contained
executable, the **bundle**, with
+the :ref:`airflow-go-pack <go-sdk/build>` tool. The bundle is the only file
you deploy.
+
+The SDK is the ``github.com/apache/airflow/go-sdk`` Go module.
+
+.. seealso::
+
+ For the full API reference and the released versions, see
+ `pkg.go.dev <https://pkg.go.dev/github.com/apache/airflow/go-sdk>`__.
.. contents:: Contents
:local:
:depth: 2
-API reference
--------------
-
-The generated API reference for the Go SDK module, and the list of its
released versions, is available on
-`pkg.go.dev <https://pkg.go.dev/github.com/apache/airflow/go-sdk>`__.
-
Prerequisites
-------------
-* Go 1.24 or later to build and pack bundles. This is a build-time requirement
only; the worker that runs a
- packed bundle needs no Go toolchain, because the bundle is a self-contained
native executable.
-* The packed bundle must be accessible from the Airflow worker and the Dag
processor, in the Dag bundle the
- coordinator scans, and built for the operating system and CPU architecture
of both.
-* The ``apache-airflow-task-sdk`` package (installed with Airflow) provides
the coordinator; no additional
- Python packages are needed.
+* Go 1.25 or later to build bundles. Airflow needs no Go toolchain to run them.
+* A bundle runs on the Airflow workers and on the Dag processor, so build it
for their operating system and CPU
+ architecture (see :ref:`go-sdk/cross-platform`).
+* The ``apache-airflow-task-sdk`` package, installed with Airflow, provides
+ :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator`, which
runs the bundle. No other Python
+ package is needed.
+* In your Go module, add the SDK, and ``airflow-go-pack`` as a tool:
-Execution architecture
-----------------------
+ .. code-block:: bash
-A Python task runner launches the Go bundle directly, with no separate Go
worker process on the host. This
-is the same coordinator mechanism the Java SDK uses. Because the mature Python
supervisor handles the
-Airflow-facing concerns, Go tasks inherit remote task logs (S3/GCS), the full
range of task states, and
-alternate XCom backends rather than implementing them again in Go.
+ go get github.com/apache/airflow/go-sdk
+ go get -tool github.com/apache/airflow/go-sdk/cmd/airflow-go-pack
.. _go-sdk/quick-start:
Quick start
-----------
-The following example shows the minimal moving parts: a Python Dag with two
stub tasks, and a Go
-implementation of those tasks.
+This example declares a whole Dag in Go, builds it into a bundle, and deploys
it, with no Python file involved.
-Python Dag (the scheduling side)
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Declare a Dag in ``main.go``:
-.. code-block:: python
+.. code-block:: go
- from airflow.sdk import dag, task
+ package main
+ import (
+ "log"
- @dag
- def simple_dag():
- @task.stub(queue="golang")
- def extract(): ...
+ "github.com/apache/airflow/go-sdk/airflow"
+ )
- @task.stub(queue="golang")
- def transform(): ...
+ func extract(airflow.Context) (int, error) {
+ return 42, nil
+ }
- extract() >> transform()
+ func report(actx airflow.Context, rows int) error {
+ actx.Logger().InfoContext(actx, "extracted rows", "rows", rows)
+ return nil
+ }
+ func main() {
+ dag := airflow.Dag("go_hello", airflow.DagSpec{Schedule: "@daily",
Queue: "golang"})
- simple_dag()
+ extracted := dag.Task(extract)
+ dag.Task(report, airflow.Inputs(extracted))
-``@task.stub`` declares the *shape* of the Go tasks (their names and
dependencies) without any Python
-implementation. The ``queue`` value routes the task to the Go coordinator.
+ bundle := airflow.Bundle()
+ bundle.Register(dag)
+ if err := bundle.Serve(); err != nil {
+ log.Fatal(err)
+ }
+ }
-Go implementation
-~~~~~~~~~~~~~~~~~~
+Build it into a bundle:
-A task is an ordinary Go function whose first parameter is an
``airflow.Context``. Everything Airflow gives
-the task is a method on it, so the signature stays the same whatever the task
uses.
+.. code-block:: bash
-.. code-block:: go
+ go tool airflow-go-pack --output dist/go_hello .
- import (
- "runtime"
+Put ``dist/go_hello`` in a Dag bundle, for example the default ``dags-folder``
bundle, which reads the
+``[core] dags_folder`` directory (see :doc:`Dag bundles
</administration-and-deployment/dag-bundles>`). Then send
+the ``golang`` queue to the Go coordinator in ``airflow.cfg``:
- "github.com/apache/airflow/go-sdk/airflow"
- )
+.. code-block:: ini
- func extract(actx airflow.Context) (any, error) {
- conn, err := actx.Client().GetConnection(actx, "test_http")
- if err != nil {
- return nil, err
- }
- actx.Logger().InfoContext(actx, "fetched connection", "host",
conn.Host)
- // ... do work, honour actx cancellation ...
- return map[string]any{"go_version": runtime.Version()}, nil
+ [sdk]
+ coordinators = {"go": {"classpath":
"airflow.sdk.coordinators.executable.ExecutableCoordinator"}}
+ queue_to_coordinator = {"golang": "go"}
+
+Airflow reads the bundle like any other Dag file: ``go_hello`` shows up in the
UI, runs on its schedule, and
+``report`` receives the value ``extract`` returned.
+
+.. _go-sdk/dag-definition:
+
+Declaring a Dag in Go
+---------------------
+
+Tasks and their inputs
+~~~~~~~~~~~~~~~~~~~~~~
+
+``airflow.Dag(dagID, spec)`` creates a Dag, and ``dag.Task(fn, opts...)`` adds
a task to it and returns a
+reference to the task. ``airflow.Inputs`` gives each parameter after the
``airflow.Context`` one input:
+
+* A task reference passes that task's return value, and makes this task wait
for it.
+* ``airflow.Literal(value)`` passes a fixed JSON value, and draws no
dependency.
+
+.. code-block:: go
+
+ type Summary struct {
+ Total int `json:"total"`
+ Regions int `json:"regions"`
}
- func transform(actx airflow.Context) error {
- val, err := actx.Client().GetVariable(actx, "my_variable")
- if err != nil {
- return err
- }
- actx.Logger().InfoContext(actx, "obtained variable", "my_variable",
val)
+ func summarize(airflow.Context) (Summary, error) {
+ return Summary{Total: 42, Regions: 2}, nil
+ }
+
+ func publish(actx airflow.Context, summary Summary, label string) error {
+ actx.Logger().InfoContext(actx, label, "total", summary.Total)
return nil
}
-.. note::
+ summarized := dag.Task(summarize)
+ dag.Task(publish, airflow.Inputs(summarized, airflow.Literal("nightly")))
+
+``publish`` receives the ``Summary`` that ``summarize`` returned, and the
literal ``"nightly"`` as ``label``.
+
+A task function takes an ``airflow.Context`` first, and returns ``error`` or
``(T, error)``. The ``T`` becomes the
+task's ``return_value`` XCom and what a downstream task receives. Values move
between tasks as JSON (see
+:ref:`go-sdk/types`), so return types JSON can hold.
+
+The inputs are checked when the Dag is built. Each parameter needs exactly one
input, an upstream task's
+result type must be assignable to the parameter type, and a literal must
decode into it. A mismatch panics,
+which the Dag processor reports as an import error.
+
+Task ids
+~~~~~~~~
- As with the other language SDKs, XCom *dependencies* are declared in the
Python stub Dag (they define task
- order). An upstream task's value reaches a downstream task either through a
parameter, when the stub Task
- passes it in the TaskFlow call (see :ref:`go-sdk/arguments`), or by reading
it explicitly with
- ``actx.Client().GetXCom``.
+The task id is the function's name, so ``dag.Task(summarize)`` adds the task
``summarize``. A method value such
+as ``svc.Extract`` takes the method's name, ``Extract``. A function literal
has no usable name, so give it one
+with ``airflow.TaskSpec{TaskID: "..."}``.
-Go entry point
-~~~~~~~~~~~~~~~
+A task id is made of letters, digits, dashes, underscores and dots, and is at
most 250 characters with the
+prefixes of its task groups. Two tasks of one Dag cannot share an id.
-Build a bundle with ``airflow.Bundle()``, register a handler for each task,
and call ``Serve`` as the last
-statement of ``main``. The ``Register`` calls are the single source of truth
for which ``dag_id`` and task
-names this bundle can run, so the generated manifest can never drift from what
the binary actually executes.
+Dag and task options
+~~~~~~~~~~~~~~~~~~~~
+
+``airflow.DagSpec`` and ``airflow.TaskSpec`` hold Airflow's own Dag and task
arguments, spelled in Go:
.. code-block:: go
- import (
- "log"
+ dag := airflow.Dag("go_etl", airflow.DagSpec{
+ Schedule: "0 6 * * *",
+ Tags: []string{"etl"},
+ Queue: "golang",
+ })
- "github.com/apache/airflow/go-sdk/airflow"
- )
+ loaded := dag.Task(loadRows, airflow.Inputs(summarized),
airflow.TaskSpec{Retries: 2})
- func main() {
- bundle := airflow.Bundle()
+``Schedule`` accepts ``@once``, ``@continuous``, a cron preset such as
``@daily``, or a cron expression. Leave it
+empty for a Dag that only runs when triggered. A cron schedule runs in UTC.
- bundle.Register(
- airflow.TaskHandler("simple_dag", "extract", extract),
- airflow.TaskHandler("simple_dag", "transform", transform),
- )
+Every task of the Dag runs in the Go bundle, so it needs a queue that
+:ref:`queue_to_coordinator <go-sdk/coordinator-config>` sends to the Go
coordinator. Set ``Queue`` once on the
+``DagSpec`` and each task inherits it. A task's own ``Queue`` wins over the
Dag's.
- if err := bundle.Serve(); err != nil {
- log.Fatal(err)
- }
+Order-only dependencies
+~~~~~~~~~~~~~~~~~~~~~~~
+
+When a task only has to run after another, without receiving its value, draw
the dependency with ``Before`` and
+``After``, the Go spelling of Python's ``>>`` and ``<<``:
+
+.. code-block:: go
+
+ cleaned := dag.Task(cleanup)
+
+ loaded.Before(cleaned) // loaded >> cleaned
+ cleaned.After(loaded, summarized) // [loaded, summarized] >>
cleaned
+ loaded.Before(airflow.Label(cleaned, "done")) // a labelled edge
+
+Both take any number of tasks or task groups, and drawing a dependency that
already exists changes nothing.
+``airflow.Label`` puts a label on the edge, which the Airflow UI shows on the
graph.
+
+Pass a value through ``airflow.Inputs`` when the downstream task needs it, and
use ``Before`` or ``After`` when it
+only needs to run in order.
+
+Task groups
+~~~~~~~~~~~
+
+``dag.TaskGroup(groupID)`` opens a group with the same ``Task``,
``TaskGroup``, ``If`` and ``Switch`` methods as
+the Dag. Every id declared in it is prefixed with the group's, as in Python:
+
+.. code-block:: go
+
+ staging := dag.TaskGroup("staging")
+ staging.Task(stageRows) // task id "staging.stageRows"
+ staging.TaskGroup("checks").Task(checkNulls) // task id
"staging.checks.checkNulls"
+
+ staging.Before(loaded) // staging >> loaded
+
+A group takes ``Before`` and ``After`` too, which orders the whole group
against a task or another group.
+
+Tasks and groups share one id namespace, so a Dag cannot hold both a task and
a group called ``staging``. Set
+``PrefixGroupID`` in an ``airflow.TaskGroupSpec`` to a pointer to ``false`` to
keep the ids declared in a group
+as written. A group id is made of letters, digits, dashes and underscores, and
is at most 200 characters.
+
+Conditional branching
+~~~~~~~~~~~~~~~~~~~~~
+
+``dag.If`` declares a task from a function that returns ``(bool, error)``, and
names the task each outcome runs:
+
+.. code-block:: go
+
+ func hasRows(_ airflow.Context, summary Summary) (bool, error) {
+ return summary.Total > 0, nil
}
-``TaskHandler`` names the ``dag_id`` and the ``task_id`` explicitly: the
``dag_id`` must match the Python
-Dag, and the ``task_id`` must match a ``@task.stub`` function in that Dag.
Neither is derived from the Go
-function name, so a handler can be named whatever reads best in Go.
+ dag.If(hasRows,
airflow.Inputs(summarized)).Then(loaded).Else(reportedEmpty)
-``TaskHandler`` also checks the signature of the function it is given and
panics if the check fails -- for
-instance when the function does not take an ``airflow.Context`` first, does
not return an ``error``, or
-declares a variadic ``...`` parameter, which no stub argument can fill.
Because ``main`` registers every
-handler before ``Serve``, a mistake stops the executable as soon as it starts
rather than when the task
-first runs.
+``If`` takes inputs and task options as ``Task`` does. ``Then`` is required,
and ``Else`` is optional: without it,
+the ``Then`` task is skipped when the condition is false.
-``Serve`` closes registration: a ``Register`` left below it in ``main`` panics
rather than adding a handler
-to the map the runtime is already answering from, so what a bundle can run
never depends on how far
-``main`` has got.
+The condition skips only the side not taken, which stays skipped if it is
cleared later. Tasks after it follow their
+own trigger rule, so with the default ``all_success``, a task after both sides
is skipped too. To run such a task when
+at least one side succeeds and none fails, give it
``airflow.TriggerRuleNoneFailedMinOneSuccess``:
-A package that defines task handlers of its own can export them as a
``[]airflow.Registerable`` for ``main``
-to pass on with ``bundle.Register(reports.Handlers()...)``.
+.. code-block:: go
-Coordinator configuration
-~~~~~~~~~~~~~~~~~~~~~~~~~~~
+ dag.Task(finish, airflow.TaskSpec{
+ TriggerRule: airflow.TriggerRuleNoneFailedMinOneSuccess,
+ }).After(loaded, reportedEmpty)
-Register a Dag bundle for the packed bundles, register the coordinator, and
route the queue to it in
-``airflow.cfg`` (or the equivalent ``AIRFLOW__*`` environment variables):
+Multi-way branching
+~~~~~~~~~~~~~~~~~~~
-.. code-block:: ini
+``dag.Switch`` declares a task from a function that returns the reference of
the task to run, and lists the
+candidates with ``Case``:
- [dag_processor]
- dag_bundle_config_list = [
- {"name": "dags-folder", "classpath":
"airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {}},
- {
- "name": "go-task-handlers",
- "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
- "kwargs": {"path": "/opt/airflow/go-task-handlers"}
+.. code-block:: go
+
+ daily := dag.Task(publishDaily)
+ weekly := dag.Task(publishWeekly)
+
+ pickCadence := func(actx airflow.Context) (*airflow.TaskRef, error) {
+ cadence, err := actx.Client().GetVariable(actx, "cadence")
+ if err != nil {
+ return nil, err
}
- ]
+ if cadence == "weekly" {
+ return weekly, nil
+ }
+ return daily, nil
+ }
- [sdk]
- coordinators = {
- "go": {
- "classpath":
"airflow.sdk.coordinators.executable.ExecutableCoordinator",
- "kwargs": {"task_handler_bundle_name": "go-task-handlers"}
- }
+ dag.Switch(pickCadence, airflow.TaskSpec{TaskID:
"pick_cadence"}).Case(daily).Case(weekly)
+
+Exactly one case runs, and the others are skipped. There is no default: a
decider that returns anything other
+than a case fails, naming what it returned and the cases it could have
returned. The deciding task's value is
+the id of the task it chose.
+
+Triggering another Dag
+~~~~~~~~~~~~~~~~~~~~~~
+
+``airflow.TriggerDagRun`` makes a task that starts a run of another Dag. Pass
it to ``dag.Task`` in place of a
+function:
+
+.. code-block:: go
+
+ triggered := dag.Task(
+ airflow.TriggerDagRun(airflow.TriggerDagRunSpec{
+ DagID: "downstream_etl",
+ Conf: map[string]any{"source": "go_etl"},
+ WaitForCompletion: true,
+ }),
+ airflow.TaskSpec{TaskID: "trigger_downstream"},
+ )
+ triggered.After(loaded)
+
+The task has no function to take an id from, so it needs a ``TaskID``, and it
takes no inputs.
+``TriggerDagRunSpec`` takes the options of Python's ``TriggerDagRunOperator``,
such as ``Conf``,
+``WaitForCompletion``, ``Deferrable``, ``PokeInterval``, ``AllowedStates``,
``FailedStates`` and
+``SkipWhenAlreadyExists``. Values are sent as written, so Jinja in ``Conf`` is
not rendered.
+
+When ``Deferrable`` is unset, it follows ``[operators] default_deferrable``. A
deferred task waits in the
+triggerer, which needs the standard provider installed. Without it,
``WaitForCompletion`` polls from the task
+and holds its worker slot.
+
+Generating Dags from data
+~~~~~~~~~~~~~~~~~~~~~~~~~
+
+A Dag is an ordinary Go value, so ``main`` can build any number of them, for
example from a loop over a
+configuration file. ``go-sdk/example/dynamic`` builds one Dag per entry of a
``dags.yaml`` file deployed next to
+the bundle, so changing the file changes the Dags with no rebuild.
+
+Registering Dags
+~~~~~~~~~~~~~~~~
+
+``bundle.Register`` takes the Dags, and ``Serve`` runs the bundle, so call it
last in ``main``. ``Register``
+checks each Dag as a whole: a cycle, an ``If`` with no ``Then``, a ``Switch``
with no ``Case``, or two Dags with
+the same ``dag_id`` panic, and the Dag processor reports an import error.
+
+A complete example
+~~~~~~~~~~~~~~~~~~
+
+``go-sdk/example/native`` uses all of the above: a task group, a fan-in, a
condition, a multi-way branch,
+order-only dependencies and a triggered Dag run.
+
+.. _go-sdk/stub-tasks:
+
+Implementing stub tasks of a Python Dag
+---------------------------------------
+
+Here a Python Dag declares the tasks and their dependencies, and Go implements
some of them. Use it when the Dag
+needs something a Go Dag cannot express yet (see :ref:`go-sdk/limitations`),
or to mix Python and Go tasks in
+one Dag.
+
+The Python Dag
+~~~~~~~~~~~~~~
+
+.. code-block:: python
+
+ from airflow.sdk import dag, task
+
+
+ @dag
+ def simple_dag():
+ @task.stub(queue="golang")
+ def extract(): ...
+
+ @task.stub(queue="golang")
+ def transform(country: str, extracted: dict): ...
+
+ transform("uk", extract())
+
+
+ simple_dag()
+
+``@task.stub`` declares a task that is implemented elsewhere, and ``queue``
sends it to the Go coordinator.
+
+The Go side
+~~~~~~~~~~~
+
+Create a ``TaskHandler`` per stub task, naming the ``dag_id`` and ``task_id``
it implements, and register them on
+the bundle:
+
+.. code-block:: go
+
+ func extract(actx airflow.Context) (any, error) {
+ return map[string]any{"rows": 42}, nil
}
- queue_to_coordinator = {"golang": "go"}
-``task_handler_bundle_name`` names the Dag bundle the coordinator scans for
packed bundles;
-``queue_to_coordinator`` routes stub tasks with ``queue="golang"`` to this Go
coordinator. See
-:ref:`go-sdk/coordinator-config` for the full list of accepted ``kwargs`` and
how bundles are located.
+ func transform(actx airflow.Context, country string, extracted
map[string]any) error {
+ actx.Logger().InfoContext(actx, "transforming", "country", country,
"rows", extracted["rows"])
+ return nil
+ }
-There is no separate Go worker to run: the Airflow worker forks the bundle
binary once per task instance.
+ func main() {
+ bundle := airflow.Bundle()
+ bundle.Register(
+ airflow.TaskHandler("simple_dag", "extract", extract),
+ airflow.TaskHandler("simple_dag", "transform", transform),
+ )
+ if err := bundle.Serve(); err != nil {
+ log.Fatal(err)
+ }
+ }
-.. note::
+The ``task_id`` must match the stub's task id, including any task group
prefix. Neither id comes from the Go
+function's name, so a handler can be named whatever reads best in Go.
- The ``[sdk]`` config and the packed bundle files must be present wherever
tasks execute and on the Dag
- processor. With ``CeleryExecutor``, tasks execute on the Celery workers;
with ``LocalExecutor``, they run
- inside the scheduler process. The Dag processor checks the stub tasks of
each Python Dag against the task
- handlers the packed bundles register, so it runs them too and needs bundles
built for its operating
- system and CPU architecture. The API server does not need any of it.
Register the Dag bundle in
- ``[dag_processor] dag_bundle_config_list`` on every component, like your
other Dag bundles: the worker
- and the Dag processor resolve ``task_handler_bundle_name`` through it, and
wherever the ``[sdk]`` config
- is read it is rejected if the name is missing there.
+``TaskHandler`` checks the function's signature and panics on a bad one, so a
mistake stops the bundle as soon
+as it starts rather than when the task first runs.
-A Dag processor with this ``[sdk]`` configuration also parses the bundle
binaries of every Dag bundle (see
-:ref:`go-sdk/native-dag-parsing`). Keep it from parsing the binaries that only
register task handlers with
-that Dag bundle's ``.airflowignore``, whose pattern syntax follows ``[core]
dag_ignore_file_syntax``. Tasks do
-not read ``.airflowignore``, so the binaries still run.
+``Register`` takes any number of task handlers and Dags, so one bundle can
implement the tasks of several Python
+Dags and declare Dags of its own. A package can export its handlers as a
``[]airflow.Registerable`` for
+``main`` to pass on with ``bundle.Register(reports.Handlers()...)``.
-With ``task_handler_bundle_name`` set to its own Dag bundle, as above,
``go-task-handlers`` holds only handler
-binaries, so ignore everything in it:
+.. _go-sdk/arguments:
-.. code-block:: bash
+TaskFlow arguments
+~~~~~~~~~~~~~~~~~~
- # [core] dag_ignore_file_syntax = glob (the default)
- echo '*' > /opt/airflow/go-task-handlers/.airflowignore
+A Python Dag that calls a stub task TaskFlow-style passes those arguments to
the handler. They bind in one of two
+ways.
- # [core] dag_ignore_file_syntax = regexp; a bare * is not a valid pattern
and is dropped
- echo '.' > /opt/airflow/go-task-handlers/.airflowignore
+**By position.** Each parameter after the ``airflow.Context`` takes the
argument in the same position, as
+``transform("uk", extract())`` fills ``country`` and ``extracted`` above.
-With ``task_handler_bundle_name`` unset, the binaries sit in the same Dag
bundle as the Python Dag file. Go
-binaries have no extension, so put them in a folder such as ``bin/`` and
ignore that folder:
+**By name.** When the handler's only parameter after the Context is a struct,
its fields take the arguments by
+name:
-.. code-block:: bash
+.. code-block:: go
- # [core] dag_ignore_file_syntax = glob (the default)
- echo 'bin/*' >> /opt/airflow/dags/.airflowignore
+ type CombineInput struct {
+ Region string `arg:"region_code"`
+ Threshold float64
+ }
- # [core] dag_ignore_file_syntax = regexp
- echo '^bin/' >> /opt/airflow/dags/.airflowignore
+ // The Python Dag calls combine(region_code="uk", threshold=0.5).
+ func combine(actx airflow.Context, input CombineInput) (any, error) {
+ return nil, nil
+ }
+
+A field takes the argument whose name matches its own, ignoring case and
underscores, so ``Threshold`` takes
+``threshold``. An ``arg:"<name>"`` tag maps a field to a different name. A
field the call does not pass keeps its
+zero value, and an argument no field takes is ignored. Both are logged as a
warning.
+
+Either way, an argument filled from another task, as ``extract()`` above,
arrives as that task's value. An
+argument the call leaves at its default needs no Go parameter, so adding a
defaulted parameter to a stub does
+not break the handler. An argument that cannot fill its Go type fails the task
before the handler runs.
+
+.. note::
+
+ A dependency drawn with ``>>`` in the Python Dag orders the tasks and passes
nothing. To read a value the
+ call did not pass, use ``actx.Client().GetXCom``.
Writing tasks
-------------
-Every task function takes an ``airflow.Context`` as its first parameter, and
reaches what Airflow provides
-through its methods:
+A task is an ordinary Go function. It works the same in both ways of using the
SDK.
+
+Its first parameter is an ``airflow.Context``, which gives it what Airflow
provides:
.. list-table::
:header-rows: 1
- :widths: 35 65
+ :widths: 25 75
* - Method
- - What it returns
+ - Returns
* - ``actx.Logger()``
- - An ``*slog.Logger`` whose output is routed back to the Airflow task log.
+ - An ``*slog.Logger`` that writes to the Airflow task log. See
:ref:`go-sdk/logging`.
* - ``actx.Client()``
- - A client for Airflow Variables, Connections, and XCom. See
:ref:`go-sdk/client`.
+ - An ``sdk.Client`` for Airflow Variables, Connections, XComs and the
task state store. See
+ :ref:`go-sdk/client`.
* - ``actx.TaskInstance()``
- - The identifiers of the running task instance. See
:ref:`go-sdk/runtime-context`.
+ - ``DagID``, ``RunID``, ``TaskID`` (including any task group prefix),
``MapIndex`` (``nil`` for an unmapped
+ task) and ``TryNumber``.
* - ``actx.DagRun()``
- - The identifiers and scheduling timestamps of its Dag run. See
:ref:`go-sdk/runtime-context`.
-
-``airflow.Context`` is itself a ``context.Context``, so pass it straight to a
client call or to
-``http.NewRequestWithContext``, and select on ``actx.Done()``, which fires
when the supervisor asks the task
-to stop. Respect it for long-running work. Cleanup that must outlive that
cancellation runs under ``context.WithoutCancel(actx)``.
-A helper typed as a plain ``context.Context`` recovers the same surface with
``airflow.FromContext``.
+ - ``DagID``, ``RunID``, and the ``*time.Time`` fields ``LogicalDate``,
``DataIntervalStart`` and
+ ``DataIntervalEnd`` (``nil`` when the run has no such value, such as a
manual run).
-Every parameter after the Context is data, filled from the stub Task's
TaskFlow call; see
-:ref:`go-sdk/arguments`.
+``airflow.Context`` is a ``context.Context``, so pass it to a client call or
to ``http.NewRequestWithContext``. Its
+``Done()`` fires when Airflow stops the task, so long-running work can stop
cleanly. Cleanup that must outlive
+that runs under ``context.WithoutCancel(actx)``. A helper that takes a plain
``context.Context`` gets the
+``airflow.Context`` back with ``airflow.FromContext``.
-An optional ``(any, error)`` return value becomes the task's ``return_value``
XCom. A non-nil ``error`` (or a
-panic, which the runtime recovers) marks the task instance failed in Airflow,
triggering retries if
-configured on the stub.
+A value the task returns becomes its ``return_value`` XCom. A non-nil
``error``, or a panic, fails the task, and
+retries it if the task has retries.
-``airflow.NewContext`` builds a Context, so a task is an ordinary function
call in a unit test:
+``airflow.NewContext`` builds a Context, so a unit test calls a task as an
ordinary function:
.. code-block:: go
@@ -292,83 +486,62 @@ configured on the stub.
airflow.TaskInstance{DagID: "simple_dag", TaskID: "transform",
TryNumber: 1},
airflow.DagRun{DagID: "simple_dag", RunID: "run1"},
)
- require.NoError(t, transform(actx))
-
-A helper the task calls can still ask for the narrowest interface it needs
(for example
-``sdk.VariableClient`` instead of the full ``sdk.Client``), which documents
the Airflow features it touches
-and lets a test pass a fake.
+ require.NoError(t, transform(actx, "uk", map[string]any{"rows": 42}))
.. _go-sdk/client:
-The ``sdk.Client`` surface
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
-``actx.Client()`` returns an ``sdk.Client``, which composes four smaller
interfaces, so a helper can depend
-on just one:
-
-* ``VariableClient`` - ``GetVariable`` (returns the Variable as a string),
``UnmarshalJSONVariable``
- (decodes a JSON Variable into a pointer you provide), ``SetVariable``, and
``DeleteVariable``.
-* ``ConnectionClient`` - ``GetConnection``, returning a ``Connection`` with
fields ``ID``, ``Type``,
- ``Host``, ``Port``, ``Login``, ``Password``, ``Path``, ``Extra`` (a
``map[string]any``), plus a
- ``GetURI()`` helper.
-* ``XComClient`` - ``GetXCom`` to read an upstream task's XCom and
``PushXCom`` to publish one.
-* ``TaskStateStoreClient`` - ``TaskStateStore``, returning the store for this
task instance:
- ``Get``, ``UnmarshalJSONValue`` (decodes a JSON value into a pointer you
provide), ``Set``,
- ``Delete``, and ``Clear``. See :ref:`go-sdk/task-state-store`.
-
-``GetXCom`` returns the stored value as an ``any``; see :ref:`go-sdk/types`
for how the stored JSON maps to
-Go types.
-
-``SetVariable`` stores the value as a string, so encode structured data (for
example with ``json.Marshal``)
-before storing it.
+The ``sdk.Client``
+~~~~~~~~~~~~~~~~~~
-.. code-block:: go
+.. list-table::
+ :header-rows: 1
+ :widths: 40 60
- client := actx.Client()
- if err := client.SetVariable(actx, "process_threshold", "42", "Rows above
this count take the slow path"); err != nil {
- return err
- }
- if err := client.DeleteVariable(actx, "legacy_threshold"); err != nil {
- return err
- }
+ * - Method
+ - Description
+ * - ``GetVariable(ctx, key)``
+ - The Variable as a string.
+ * - ``UnmarshalJSONVariable(ctx, key, pointer)``
+ - Decodes a JSON Variable into the pointer.
+ * - ``SetVariable(ctx, key, value, description)``
+ - Stores a Variable, replacing any existing value. Values are strings, so
encode structured data, for
+ example with ``json.Marshal``.
+ * - ``DeleteVariable(ctx, key)``
+ - Removes a Variable.
+ * - ``GetConnection(ctx, connID)``
+ - A ``Connection`` with ``ID``, ``Type``, ``Host``, ``Port``, ``Login``,
``Password``, ``Path`` and ``Extra``,
+ and a ``GetURI()`` helper.
+ * - ``GetXCom(ctx, dagID, runID, taskID, mapIndex, key, nil)``
+ - An XCom value, as an ``any``. See :ref:`go-sdk/types`.
+ * - ``PushXCom(ctx, ti, key, value)``
+ - Publishes an XCom value.
+ * - ``TaskStateStore()``
+ - This task instance's key/value store. See
:ref:`go-sdk/task-state-store`.
+
+A missing value returns an error that wraps ``sdk.VariableNotFound``,
``sdk.ConnectionNotFound``,
+``sdk.XComNotFound`` or ``sdk.TaskStateNotFound``, so test for it with
``errors.Is``.
+
+``sdk.Client`` is made of ``VariableClient``, ``ConnectionClient``,
``XComClient`` and ``TaskStateStoreClient``,
+so a helper can take only the one it needs, and a test can pass a fake.
.. note::
- A value supplied by a secrets backend (for example an ``AIRFLOW_VAR_*``
environment variable) still takes
- precedence over the stored value when the Variable is read back. Calling
``SetVariable`` with an empty
- description clears any existing description.
-
-Not-found lookups return sentinel errors - ``VariableNotFound``,
``ConnectionNotFound``, ``XComNotFound``,
-``TaskStateNotFound`` - so you can branch on a missing value with
``errors.Is`` rather than parsing an error
-string.
+ A value from a secrets backend, such as an ``AIRFLOW_VAR_*`` environment
variable, still wins over the
+ stored value when the Variable is read back. ``SetVariable`` with an empty
description clears the existing
+ description.
.. _go-sdk/task-state-store:
-The task state store
-~~~~~~~~~~~~~~~~~~~~~~
-
-``actx.Client().TaskStateStore()`` returns a persistent key/value store
private to one task instance,
-and the Go SDK's entry point to durable execution. It is the same store the
Python SDK exposes as
-``context["task_state_store"]``; see :doc:`/core-concepts/task-state-store`
for the concept and its
-configuration.
+Task state store
+~~~~~~~~~~~~~~~~
-The store is scoped to ``dag_id``, ``run_id``, ``task_id``, and ``map_index``.
It deliberately does *not*
-include ``try_number``, so a value written by one attempt is still readable by
the next one: a task that
-records an external job ID or its own progress can resume after a worker crash
or a retry instead of
-redoing the work. The Execution API confines every call to the task instance
the caller is running as, so
-there is no way to address another task's store - pass results between tasks
with XCom instead.
-
-The usual shape is to look for a checkpoint first and only do the expensive
work when it is missing:
+The task state store is a per-task-instance key/value store, scoped to the
Dag, run, task and map index, but not
+the try number, so a value written by one attempt is still readable by the
next. It survives worker crashes and
+retries, which makes it a good place to record an external job id or a
checkpoint. See
+:doc:`/core-concepts/task-state-store` for the concept.
.. code-block:: go
- import (
- "errors"
-
- "github.com/apache/airflow/go-sdk/airflow"
- "github.com/apache/airflow/go-sdk/sdk"
- )
-
func runSparkJob(actx airflow.Context) error {
store := actx.Client().TaskStateStore()
@@ -376,8 +549,7 @@ The usual shape is to look for a checkpoint first and only
do the expensive work
stored, err := store.Get(actx, "job_id")
switch {
case errors.Is(err, sdk.TaskStateNotFound):
- // First attempt: submit the job and remember its ID before doing
anything else.
- if jobID, err = sparkClient.SubmitJob(actx); err != nil {
+ if jobID, err = submitSparkJob(actx); err != nil {
return err
}
if err := store.Set(actx, "job_id", jobID,
sdk.WithRetention(sdk.NeverExpire)); err != nil {
@@ -386,136 +558,44 @@ The usual shape is to look for a checkpoint first and
only do the expensive work
case err != nil:
return err
default:
- // Get returns an any; the value was stored by this task as a
string.
jobID = stored.(string)
- actx.Logger().InfoContext(actx, "reattaching to job submitted by
an earlier attempt", "job_id", jobID)
}
- return sparkClient.WaitForCompletion(actx, jobID)
+ return waitForSparkJob(actx, jobID)
}
-``value`` must not be nil and must be JSON-representable - a string, number,
bool, slice, map, or a struct,
-which is stored as an object built from its exported fields and their ``json``
tags. A custom
-``MarshalJSON`` is not called, so a type that relies on one is stored as the
shape of its fields, and a
-struct with no exported fields is stored as ``{}``. Read a scalar back with
``Get``, which returns it as an ``any`` (the
-numeric caveat in :ref:`go-sdk/types` applies here too); for an object or
array,
-``UnmarshalJSONValue`` decodes it straight into a pointer you provide.
-
-A value the store cannot hold is rejected before it is sent, so you get an
error naming the problem rather
-than a round trip that fails on the server. The one that catches people out is
``time.Time``, which JSON has
-no spelling for - store ``value.Format(time.RFC3339)`` and parse it back with
``time.Parse``. Non-finite
-floats and ``[]byte`` are refused for the same reason. This mirrors the Python
SDK, where the same values
-fail Pydantic validation before the write leaves the worker.
-
-Keys expire, so retention is part of writing a value:
-
-* ``Set`` without options uses the deployment's ``[state_store]
default_retention_days`` (30 days by
- default). The Go runtime cannot read Airflow's config, so the supervisor
resolves that value and passes
- it in the environment when it launches the bundle. A deployment that sets it
to something unusable - a
- negative number, or a value that is not a whole number of days - fails the
write, exactly as it does for
- a Python task, rather than quietly substituting a different lifetime.
-* ``sdk.WithRetention`` takes an explicit, positive ``time.Duration``, or
``sdk.NeverExpire`` for a key
- that is skipped by garbage collection entirely. A zero or negative retention
is rejected rather than
- given a meaning of its own: to follow the deployment default omit the
option, and to drop a key call
- ``Delete``.
-
-``Delete`` removes one key (deleting a key that does not exist is not an
error) and ``Clear`` removes
-every key stored for this task instance.
-
-.. note::
+The store has ``Get``, ``UnmarshalJSONValue`` (decodes an object or array into
a pointer), ``Set``, ``Delete``
+and ``Clear``.
- The Go SDK does not implement the worker-side state backend (``[workers]
state_store_backend``), which
- offloads large values to external storage and records only a reference
marker in the database. If a
- deployment configures one, a Go task reading a key that was written through
that backend receives the raw
- reference marker rather than the original value, and a Go task writing a key
stores the whole value in the
- database instead of offloading it. This is the same behaviour as a Python
worker that does not have the
- backend configured.
+* A value must be JSON, and not ``nil``. A ``time.Time``, a non-finite float
or a ``[]byte`` is refused before it
+ is sent, so store a time as ``value.Format(time.RFC3339)``.
+* ``Set`` without options keeps the key for ``[state_store]
default_retention_days``.
+ ``sdk.WithRetention(d)`` keeps it for a positive ``time.Duration``, and
``sdk.NeverExpire`` keeps it forever.
+* ``[workers] state_store_backend`` is not used from Go tasks, so values are
always stored in the database. A
+ key a Python task wrote through such a backend reads back as the backend's
reference marker.
-.. _go-sdk/runtime-context:
+.. _go-sdk/logging:
-Reading the task runtime context
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Logging
+~~~~~~~
-``airflow.Context`` carries the identifiers and scheduling timestamps of the
running task instance and its
-Dag run -- the Go equivalent of the execution context the Python and Java SDKs
expose:
+``actx.Logger()`` writes to the Airflow task log, at the level ``[logging]
logging_level`` sets.
+``[logging] namespace_levels`` applies to a logger named with ``WithGroup``:
.. code-block:: go
- func extract(actx airflow.Context) (any, error) {
- ti := actx.TaskInstance()
- actx.Logger().InfoContext(actx, "running",
- "dag_id", ti.DagID,
- "run_id", ti.RunID,
- "task_id", ti.TaskID,
- "try_number", ti.TryNumber,
- "logical_date", actx.DagRun().LogicalDate,
- )
- return nil, nil
- }
-
-``actx.TaskInstance()`` returns ``DagID``, ``RunID``, ``TaskID``, ``MapIndex``
(nil for an unmapped task),
-and ``TryNumber``; ``actx.DagRun()`` returns ``DagID``, ``RunID``, and the
``*time.Time`` fields
-``LogicalDate``, ``DataIntervalStart``, and ``DataIntervalEnd`` (nil when the
run has no such value, e.g. a
-manual trigger).
-
-.. _go-sdk/arguments:
-
-Receiving arguments from the stub Task
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
-A stub Task's arguments reach a Go handler in one of two ways:
-
-1. **Positional binding** -- each data parameter takes the argument in the
same position.
-2. **Struct-based (keyword) binding** -- a sole struct parameter takes the
arguments by field name.
-
-Every parameter after the ``airflow.Context`` is a **data parameter**, filled
in declaration order from the
-arguments of the Python stub Task's TaskFlow call. A literal in the Dag file
(``transform("uk", ...)``)
-decodes straight into the parameter; an upstream task's output
(``transform(..., extract())``) is pulled
-from that task's XCom in the current Dag run. If the argument count does not
match, or an argument's
-declared type cannot fill the Go type, the task fails before its body runs.
-
-.. code-block:: go
-
- // The Python stub Task calls transform("uk", extract()).
- func transform(actx airflow.Context, country string, extracted
map[string]any) error {
- actx.Logger().InfoContext(actx, "transforming", "country", country)
- return nil
- }
-
-When a task's **sole** data parameter is a struct, its fields bind **by name**
instead of by position --
-keyword arguments rather than positional ones. Being the only data parameter
is the opt-in; there is no
-marker to add.
-
-.. code-block:: go
-
- type CombineInput struct {
- Region string `arg:"region_code"` // renamed
- Threshold float64
- }
+ databaseLog := actx.Logger().WithGroup("example.database")
+ databaseLog.Debug("query complete", "rows", 42)
- // The Python stub Task calls combine(region_code="uk", threshold=0.5).
- func Combine(actx airflow.Context, input CombineInput) (any, error) {
- return nil, nil
- }
-
-An exported field binds the argument matching its own Go name, folding case
and underscores, so
-``Threshold`` takes ``threshold``; reach for an ``arg:"<name>"`` tag when the
names genuinely differ, as
-``Region`` does above. The `Go SDK README
-<https://github.com/apache/airflow/blob/main/go-sdk/README.md>`__ has the full
binding rules, including
-how unmatched fields and arguments are treated and when an untagged struct is
decoded whole from a single
-argument instead.
-
-Stub parameters the Dag author left at their Python defaults are the exception
to both shapes: they reach
-the wire but need no Go parameter, so adding a defaulted parameter to a stub
does not break the Go
-functions already bound to it.
+A namespace level applies to the longest matching dotted prefix, so
``example=DEBUG`` also applies to
+``example.database``.
.. _go-sdk/types:
XCom type mapping
-----------------
-XCom values are stored as JSON in Airflow's metadata database. The table below
shows how those JSON types
-surface as Go values when read back via ``GetXCom``.
+XComs are stored as JSON, so a value read with ``GetXCom`` arrives as the
matching Go type:
.. list-table::
:header-rows: 1
@@ -523,10 +603,10 @@ surface as Go values when read back via ``GetXCom``.
* - Python type
- JSON
- - Go type (from ``GetXCom``)
+ - Go type
* - ``int``
- number (integer)
- - numeric (see note)
+ - an integer type (see note)
* - ``float``
- number (decimal)
- ``float64``
@@ -548,123 +628,148 @@ surface as Go values when read back via ``GetXCom``.
.. note::
- ``GetXCom`` returns the value exactly as decoded from the transport; there
is no typed XCom
- deserialization layer yet. The Python supervisor encodes values as
``msgpack``, so a whole number arrives
- as a Go integer type (whose width depends on the value) and only a
non-integer arrives as ``float64``. Do
- not assume a fixed integer width: type-switch over the numeric types you
expect, or round-trip the value
- through ``json.Marshal`` / ``json.Unmarshal`` into a typed Go value.
+ An integer arrives as a Go integer type whose width depends on the value.
Type-switch over the integer types
+ you expect, or round-trip the value through ``json.Marshal`` and
``json.Unmarshal`` into a typed Go value. A
+ task input from ``airflow.Inputs`` or a TaskFlow argument is already decoded
into its parameter type.
.. _go-sdk/build:
-Building and packaging
-----------------------
-
-A plain ``go build`` produces a runnable binary, but a *deployable* bundle
(binary + embedded source files +
-manifest) must be produced with ``airflow-go-pack``. The packer compiles the
bundle and appends the embedded
-metadata footer, so the coordinator can read its ``dag_id``\ s without
executing the binary, producing a
-single runnable file. The on-disk format the packer emits (the ``AFBNDL01``
footer and the
-``airflow-metadata.yaml`` manifest) is the bundle format shared by all
native-executable SDKs, specified in
-:doc:`task-sdk:executable-bundle-spec`.
-
-``airflow-go-pack`` ships via the Go 1.24 ``tool`` directive, so there is no
global install: add
-
-.. code-block:: text
-
- tool github.com/apache/airflow/go-sdk/cmd/airflow-go-pack
-
-to your bundle module's ``go.mod`` and run it with ``go tool
airflow-go-pack``. This pins the packer version
-per project.
+Building a bundle
+-----------------
-Build and pack in one step; any flags after ``--`` are forwarded verbatim to
``go build``:
+``airflow-go-pack`` builds your ``main`` package into a single executable, the
bundle. It also embeds the
+entrypoint file and the file that declares each Dag, which the Airflow UI's
**Code** view shows, and a manifest
+of the bundle's Dags and task handlers. That file is all you deploy, and
Airflow refuses to run a bundle whose
+content was changed after it was built.
.. code-block:: bash
- go tool airflow-go-pack ./example/bundle -- -trimpath -tags=prod
+ go tool airflow-go-pack --output dist/my_dags ./cmd/my_dags -- -trimpath
-Use ``--output <path>`` to write the packed bundle straight into the directory
of the Dag bundle the
-coordinator scans (see `Deploying`_):
+* ``--output <path>`` sets the output file. By default, the bundle is written
to the current directory, named
+ after the package's directory.
+* Flags after ``--`` are passed to ``go build``.
+* ``go tool airflow-go-pack inspect <bundle>`` prints a bundle's manifest.
-.. code-block:: bash
+The Dag processor reads a bundle only when its file name has no extension,
such as ``dags/orders``, not
+``dags/orders.bin``.
- go tool airflow-go-pack --output
/opt/airflow/go-task-handlers/sample-dag-bundle ./example/bundle
+.. _go-sdk/cross-platform:
Cross-platform builds
-~~~~~~~~~~~~~~~~~~~~~~~
+~~~~~~~~~~~~~~~~~~~~~
-The worker and the Dag processor that run a bundle often use a different
operating system or CPU
-architecture than your build machine (for example, deploying to a Linux host
from an Apple-silicon
-``darwin/arm64`` laptop). Pass ``--goos`` / ``--goarch`` and the packer
cross-builds for you:
+The workers and the Dag processor often run a different operating system or
CPU architecture than your build
+machine. Pass ``--goos`` and ``--goarch`` to build for them:
.. code-block:: bash
- go tool airflow-go-pack --goos linux --goarch amd64 \
- --output /opt/airflow/go-task-handlers/sample-dag-bundle \
- ./example/bundle
+ go tool airflow-go-pack --goos linux --goarch amd64 --output dist/my_dags
./cmd/my_dags
-Alternatively, pack a pre-built binary with ``--executable`` / ``--source``.
The packer normally execs the
-binary with ``--airflow-metadata`` to read its manifest, but a cross-compiled
binary cannot run on the build
-host. In that case, generate the manifest on a machine that *can* run the
binary and feed it to the packer
-with ``--airflow-metadata``:
+``airflow-go-pack`` runs the built binary to read its manifest. When it cannot
do that for a cross-compiled
+binary, build and read the manifest on a machine of the target platform, then
pack the binary with
+``--executable``:
.. code-block:: bash
# On a linux/amd64 machine:
- go build -o my-bundle ./example/bundle
- ./my-bundle --airflow-metadata > airflow-metadata.yaml
+ go build -o my_dags ./cmd/my_dags
+ ./my_dags --airflow-metadata > airflow-metadata.yaml
- # Back on the darwin/arm64 machine:
- go tool airflow-go-pack --executable ./my-bundle --source main.go \
- --airflow-metadata airflow-metadata.yaml
-
-(``--executable`` is mutually exclusive with ``--goos`` / ``--goarch`` and
with ``go build`` flags after
-``--``, since it packs an already-built binary instead of building one.)
+ # On the build machine:
+ go tool airflow-go-pack --executable ./my_dags --source
cmd/my_dags/main.go \
+ --airflow-metadata airflow-metadata.yaml --output dist/my_dags
Deploying
-~~~~~~~~~
+---------
-Copy or mount the packed bundle into the Dag bundle named by the coordinator's
``task_handler_bundle_name``.
-The :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator` scans
that Dag bundle recursively,
-matches the incoming ``dag_id`` against each bundle's manifest, verifies the
bundle's integrity hash, and
-launches the matching bundle. This scan identifies bundles by the trailer
magic, not by file name. The Dag
-processor also requires a file name without an extension, see
:ref:`go-sdk/native-dag-parsing`.
+Where the bundle goes depends on what it contains.
-The matching bundle is marked executable before it is launched, so any Dag
bundle works, including an
-object-store one such as ``S3DagBundle`` that has no concept of file
permissions and so cannot preserve
-the execute bit the build produced.
+Dag definition
+~~~~~~~~~~~~~~
-.. _go-sdk/native-dag-parsing:
+Put the bundle in a Dag bundle, such as the default ``dags-folder`` bundle.
The Dag processor runs each bundle it
+finds to read its Dags, as it reads any other Dag file, so it needs the same
``[sdk]`` configuration as the workers,
+shown in the :ref:`quick start <go-sdk/quick-start>`, and a bundle built for
its platform.
-Parsing native Dags
-~~~~~~~~~~~~~~~~~~~
+With more than one ``ExecutableCoordinator``, map each Dag bundle that holds
Go Dags to one of them in
+``[sdk] dag_bundle_to_coordinator``, such as ``{"dags-folder": "go"}``, or its
bundles fail to parse.
-Once an :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator` is
configured, the Dag processor
-parses every bundle binary in each Dag bundle and runs it to collect the Dags
it defines. The Dag processor
-recognizes a bundle binary only when its file name has no extension, such as
``bin/orders``, not
-``bin/orders.bin``. A binary that only registers task handlers is parsed too:
each parse runs it and finds
-no Dags, and a binary built with a Go SDK that cannot answer the Dag-parse
request records an import error.
-Keep those out of the Dag processor with ``.airflowignore``, as described in
:ref:`Quick start <go-sdk/quick-start>`.
+Their tasks run from the same Dag bundle, at the version their Dag run was
created with, so a Dag declared in Go
+needs no coordinator option to find its bundle.
-With one :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator`,
it parses the binaries of every Dag
-bundle. With several, for example a second one for another
``task_handler_bundle_name``, map each Dag bundle that
-holds native Go Dags to one of them in ``[sdk] dag_bundle_to_coordinator``. A
binary in a Dag bundle that has no
-entry, or whose entry names no
:class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator`, fails to
parse
-with an import error:
+A bundle that cannot be read, or whose Dags fail a check, shows up as an
import error, and the other Dags keep
+working.
+
+Python Dag with stub TaskHandler
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+By default, the coordinator loads the bundle that implements a stub task from
the stub task's own Dag bundle, at
+the version its Dag run was created with, so the bundle goes next to the
Python Dag that declares the stub.
+
+To keep such bundles in a Dag bundle of their own, name it with
``task_handler_bundle_name``. A task then uses the
+version that bundle is on when the task starts:
.. code-block:: ini
+ [dag_processor]
+ dag_bundle_config_list = [
+ {"name": "dags-folder", "classpath":
"airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {}},
+ {
+ "name": "go-task-handlers",
+ "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
+ "kwargs": {"path": "/opt/airflow/go-task-handlers"}
+ }
+ ]
+
[sdk]
- dag_bundle_to_coordinator = {"dags-folder": "go"}
+ coordinators = {
+ "go": {
+ "classpath":
"airflow.sdk.coordinators.executable.ExecutableCoordinator",
+ "kwargs": {"task_handler_bundle_name": "go-task-handlers"}
+ }
+ }
+ queue_to_coordinator = {"golang": "go"}
+
+The coordinator finds the bundle for a task by the ``dag_id`` in each bundle's
manifest, not by file name, so one
+Dag bundle can hold several bundles. Any Dag bundle type works, including an
object-store one such as
+``S3DagBundle``.
+
+The Dag processor checks the stub tasks of each Python Dag against the task
handlers the bundles register, so
+it needs the bundles too, built for its platform.
+
+The Dag processor also parses a bundle that only implements stub tasks, which
runs it once per parse and finds
+no Dags. Skip that with the Dag bundle's ``.airflowignore``; the coordinator
still finds the bundles. For a Dag
+bundle that holds only such bundles, as ``go-task-handlers`` above:
+
+.. code-block:: bash
+
+ # [core] dag_ignore_file_syntax = glob (the default)
+ echo '*' > /opt/airflow/go-task-handlers/.airflowignore
+
+ # [core] dag_ignore_file_syntax = regexp
+ echo '.' > /opt/airflow/go-task-handlers/.airflowignore
+
+For bundles next to a Python Dag, put them in a folder such as ``bin/`` and
ignore it with ``bin/*`` (glob) or
+``^bin/`` (regexp).
-* The Code view shows each native Dag's own source file, taken from the
sources the bundle embeds.
-* A task of a native Go Dag runs the bundle binary its Dag was parsed from.
+Where the configuration is needed
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The coordinator runs where tasks run, so the ``[sdk]`` configuration and the
bundles are needed on the workers.
+With ``CeleryExecutor``, set them on the Celery workers. With
``LocalExecutor``, tasks run in subprocesses on the
+scheduler's host, so set them there. The Dag processor needs them too, as
described above. Register every Dag bundle in
+``[dag_processor] dag_bundle_config_list`` on each of these components. The
API server never needs them.
+
+There is no separate Go service to run: the worker starts the bundle once per
task instance.
.. _go-sdk/coordinator-config:
:class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator`
configuration
----------------------------------------------------------------------------------------------
+---------------------------------------------------------------------------------
-All ``kwargs`` in the ``coordinators`` config entry are passed to the
-:class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator`
constructor:
+The ``kwargs`` of a ``coordinators`` entry are passed to
+:class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator`, and all
of them are optional:
.. list-table::
:header-rows: 1
@@ -674,33 +779,29 @@ All ``kwargs`` in the ``coordinators`` config entry are
passed to the
- Default
- Description
* - ``task_handler_bundle_name``
- - *(task's own Dag bundle)*
- - Name of the Dag bundle scanned recursively for executable bundles. It
is used only by
- mixed-language Dags, to locate the task handlers for the ``@task.stub``
tasks of a Python Dag;
- Dags defined natively in a language SDK do not use it. It must be
registered in
- ``[dag_processor] dag_bundle_config_list``. It is checked when the
``[sdk]`` configuration is
- loaded, so a typo fails there rather than on the first task.
+ - *(unset)*
+ - The Dag bundle to load the bundles that implement stub tasks from. When
unset, a stub task loads its
+ bundle from its own Dag bundle. It must be registered in
``[dag_processor] dag_bundle_config_list``.
* - ``task_startup_timeout``
- ``10.0``
- - Seconds to wait for the bundle subprocess to connect after launch.
Increase this if your
- bundle startup is slow (e.g. on constrained hardware).
-
-.. note::
+ - Seconds to wait for a task's bundle to start. Increase it if your
bundle starts slowly, for example on
+ constrained hardware.
- **Locating bundles.** The packed bundles for the ``@task.stub`` tasks of a
Python Dag are read from a Dag
- bundle, so they are delivered, refreshed and versioned by the same machinery
as your Dags.
+``task_handler_bundle_name`` only concerns stub tasks: a Dag declared in Go
always runs from the Dag bundle it was
+read from.
- * The expected layout is a separate Dag bundle for the packed bundles, named
by
- ``task_handler_bundle_name``, rather than the Dag bundle that holds your
``.py`` files. The task
- uses the version that Dag bundle is on when it starts, pinned for the
whole task.
- * If ``task_handler_bundle_name`` is unset, packed bundles are read from the
**task's own** Dag
- bundle, pinned to the version the run was created with.
+``queue_to_coordinator`` maps a queue to a coordinator entry, so every task on
that queue runs in a Go bundle.
.. _go-sdk/limitations:
Limitations
-----------
-* **A Python stub Dag is still required.** The Execution API does not yet
carry Dag structure for non-Python
- languages, so task names and dependencies are declared in Python with
- :func:`@task.stub <airflow.sdk.task.stub>`. This is a documented known
limitation.
+* **A Dag declared in Go cannot express everything yet.** Assets, Dag params,
custom timetables, dynamic task
+ mapping, setup and teardown tasks, and human-in-the-loop tasks have no Go
form. Declare such a Dag in Python,
+ and implement its tasks in Go with :func:`@task.stub
<airflow.sdk.task.stub>`.
+* **Cluster policies do not apply to Dags declared in Go.** ``dag_policy`` and
``task_policy`` do not run when the
+ Dag processor reads them.
+* **Some CLI commands do not take a Dag declared in Go.** ``airflow dags
test``, ``tasks test`` and
+ ``tasks render`` refuse it, and ``airflow dags reserialize`` skips it.
+* **One process per task instance.** Tasks cannot share in-process state; use
XComs or an external store.
diff --git a/go-sdk/DEVELOPMENT.md b/go-sdk/DEVELOPMENT.md
new file mode 100644
index 00000000000..b6511e8ec99
--- /dev/null
+++ b/go-sdk/DEVELOPMENT.md
@@ -0,0 +1,215 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+ -->
+
+# Developing the Go SDK
+
+This guide is for contributors to the Go SDK. To use the SDK, see the
[README](./README.md).
+
+## Building and testing
+
+The [`Justfile`](./Justfile) has the common tasks:
+
+```bash
+just test # run the unit tests
+just build # pack the example bundles
+just docs # serve the package docs locally
+just generate-models # regenerate the coordinator-protocol models
+just generate-specs # regenerate the Dag, task and task group specs
+```
+
+## How it works
+
+The bundle binary speaks the coordinator protocol when Airflow launches it
with `--comm` and `--logs`.
+
+```
+Python supervisor / task runner
+ │ finds + validates the bundle, then forks it:
+ ▼
+ <bundle binary> --comm=127.0.0.1:P1 --logs=127.0.0.1:P2
+ │ binary dials BACK over TCP loopback (msgpack-over-IPC)
+ ▼
+ GetConnection / GetVariable / GetXCom / SetXCom ... → SucceedTask / TaskState
+```
+
+- The Python `ExecutableCoordinator` forks the bundle binary with
`--comm`/`--logs` addresses it is already
+ listening on. The binary dials back (it never listens) and speaks a
length-prefixed msgpack-over-IPC wire
+ protocol on the comm socket, with structured JSON-line logs on the logs
socket.
+- The Python runtime is the worker. It proxies every `GetConnection` /
`GetVariable` / `GetXCom` /
+ `SetXCom` call through to the Execution API. The Go binary just runs the
task function.
+- When the first frame on the comm socket is a `DagFileParseRequest` from the
Dag processor, the
+ binary runs no task. It answers with one `DagFileParsingResult` that holds
the Dags from
+ `airflow.Dag` that the binary registered. The Dags are serialized as
+ [Serializing a native Dag](#serializing-a-native-dag) describes. The binary
then exits.
+
+The Go side of the protocol is implemented in `pkg/execution/`. On the Python
side it is the
+`ExecutableCoordinator` in
`task-sdk/src/airflow/sdk/coordinators/executable/coordinator.py`.
+
+## The vendored schemas
+
+`schema/dag-schema.json` and `schema/supervisor-schema.json` are byte-for-byte
copies of schemas
+airflow-core and task-sdk own. The generators read the copies, not the
originals, so a generated file
+is explainable from a file inside this module and a standalone checkout — a
published Go module, an
+ASF source release — can regenerate and test without the monorepo around it.
`ts-sdk` and `java-sdk`
+vendor theirs the same way.
+
+Two prek hooks split the work that keeps the copies honest.
`sync-go-sdk-schemas` copies an original
+over its copy when the two differ and fails, since copying is mechanical; it
is a **manual** hook, so
+re-vendoring is a deliberate step here rather than something a change to a
Python schema forces on
+whoever made it — which also means nothing announces on its own that a copy
has gone stale.
+`check-go-sdk-generated-drift` runs on every commit and regenerates from the
copies, failing when the
+committed Go differs, since what to do about a new schema construct — a
generator rule, an authoring
+exclusion — is a decision.
+
+## Regenerating the coordinator-protocol models
+
+The types in `pkg/execution/genmodels/` are generated from
`schema/supervisor-schema.json`, this
+module's vendored copy of the supervisor schema snapshot the Python Task SDK
owns
+(`task-sdk/src/airflow/sdk/execution_time/schema/schema.json`); do not edit
either by hand. To move
+the SDK to a newer schema version:
+
+1. Refresh the copy: `prek run sync-go-sdk-schemas --hook-stage manual`, which
overwrites
+ `schema/supervisor-schema.json` from task-sdk's and fails so the change
lands in review.
+2. Set `SupervisorSchemaVersion` in
[`pkg/execution/messages.go`](./pkg/execution/messages.go) to the
+ copy's `api_version` date.
+3. Run `just generate-models`, and commit the copy with what it generated.
+
+`TestSupervisorSchemaVersionMatchesSnapshot` fails when the constant and the
copy's `api_version`
+drift, so a missed bump is caught by `go test`. A snapshot can also grow a
field without the
+`api_version` moving, which leaves the models behind with nothing failing —
msgpack drops a field the
+struct does not declare — so the `check-go-sdk-generated-drift` prek hook
regenerates them and fails
+when the committed files differ.
+
+[`airflow/enums.go`](./airflow/enums.go) declares the `genmodels.DagRunState`
constants again, so that
+a Dag author does not import `genmodels`. `TestDagRunStateMatchesGenmodels`
fails when regenerating
+the models adds, renames or removes a `DagRunState` constant. The test keeps
failing until
+`airflow/enums.go` declares the same constants as `genmodels`.
+
+## Regenerating the Dag, task and task group specs
+
+`airflow.DagSpec`, `airflow.TaskSpec` and `airflow.TaskGroupSpec` in
+[`airflow/spec.gen.go`](./airflow/spec.gen.go) are generated from
`schema/dag-schema.json`, this
+module's vendored copy of airflow-core's Dag serialization schema
+(`airflow-core/src/airflow/serialization/schema.json`), which Python owns; do
not edit any of them
+by hand. Refresh the copy with `prek run sync-go-sdk-schemas --hook-stage
manual`,
+then run `just generate-specs` after changing the schema or the generator.
+
+The schema is the serialized shape rather than the authoring one, so
+[`internal/genspec/authoring.go`](./internal/genspec/authoring.go) holds the
three tables that turn
+it into the authoring shape, each entry carrying the reason it exists:
+
+- **exclusions** — properties an author never sets, such as the paths the
bundle fills in and the
+ template fields of a Python operator class. A property *not* excluded
generates, so one added on
+ the Python side surfaces in review instead of vanishing.
+- **type overrides** — the schema types a moment in time and a duration as a
number of seconds, and
+ an integral count as a JSON number.
+- **injections** — `Schedule`, which stands in for the serialized `timetable`,
and the `Queue` of
+ `DagSpec`, for which the schema has no Dag-level property. Injecting into
the schema rather than
+ hand-writing the field keeps every field in one struct declaration, which is
what lets
+ `TaskSpec` implement the sealed `TaskOption`.
+
+The schema types `trigger_rule` and `weight_rule` as plain strings and does
not list their values.
+`TriggerRule`, `WeightRule` and their constants are therefore hand-written in
+[`airflow/enums.go`](./airflow/enums.go). `TestRuleConstantsMatchPython`
checks those constants
+against the Python enums in airflow-core.
+
+`just generate-specs` also writes
[`airflow/spec_fields.gen.go`](./airflow/spec_fields.gen.go), which
+holds the schema key and the schema default of each field, keyed by the name
of the Go field.
+
+The `check-go-sdk-generated-drift` prek hook regenerates both files and fails
when a committed one
+differs, so a schema change that never reached Go cannot merge.
+
+## Serializing a native Dag
+
+[`airflow/serialize.go`](./airflow/serialize.go) writes a registered
`airflow.Dag` as the Dag JSON
+that Airflow stores. Python's `DagSerialization` writes the same shape for a
Python Dag. Each field
+of `DagSpec`, `TaskSpec` and `TaskGroupSpec` has an entry in a generated
table, which names the
+schema property that the field sets and its schema default. The serializer
leaves out a value equal to
+the default. `serialize.go` lists only the fields that the generated tables
cannot describe: the ones
+it writes in its own way, and the lists that Python keeps in a set.
+
+`airflow.Inputs` takes the results of tasks and `airflow.Literal` values, one
for each parameter after
+the `Context`. A task becomes an `xcom` argument binding and a literal becomes
a `literal` one:
+
+```go
+dag.Task(load, airflow.Inputs(transformed, airflow.Literal("s3://bucket/out")))
+```
+
+```json
+[
+ {"name": "arg0", "kind": "xcom", "task_id": "transform"},
+ {"name": "arg1", "kind": "literal", "value": "s3://bucket/out"}
+]
+```
+
+A literal must be JSON that decodes into its parameter type, and it adds no
edge to the Dag.
+
+The `check-go-sdk-serialization-conformance` prek hook builds the Dags of
+[`scripts/ci/lang_sdk_serialization/test_dags.yaml`](../scripts/ci/lang_sdk_serialization/test_dags.yaml)
+twice: with this SDK, and with Python and Airflow's own serializer. It loads
the Go output with
+Airflow's deserializer and compares the two serializations field by field. Run
it from the
+repository root:
+
+```bash
+prek run check-go-sdk-serialization-conformance --all-files
+```
+
+The hook runs every Dag of the file, including the ones that `requires` a
feature that only some SDKs
+have.
[`go_native.json`](../airflow-core/tests/unit/dag_processing/lang_sdk_fixtures/go_native.json)
+holds the Go output for the same Dags, so that the Airflow core tests check
that Airflow loads it.
+`TestSerializeConformanceDags` says how to rewrite it.
+
+## Architectural decisions
+
+The [`adr/`](./adr) directory records the design decisions behind the SDK:
+
+- [ADR 0001](./adr/0001-bundle-packing-options.md): bundle-packing options.
+- [ADR 0002](./adr/0002-use-go-tool-directive-for-bundle-packer.md): deliver
the bundle packer via the
+ Go 1.24 `tool` directive.
+- [ADR 0003](./adr/0003-coordinator-protocol-msgpack-ipc.md): the
msgpack-over-IPC coordinator protocol.
+- [ADR 0004](./adr/0004-self-contained-executable-bundle.md): the
self-contained executable bundle, where
+ the executable *is* the bundle.
+- [ADR 0005](./adr/0005-retire-go-edge-worker.md): retire the standalone Go
Edge Worker and make the
+ coordinator the only execution path.
+- [ADR 0006](./adr/0006-cross-language-argument-binding.md): cross-language
TaskFlow argument binding,
+ from serialized literals and XCom references to typed Go parameters.
+- [ADR 0007](./adr/0007-mixed-lang-task-handler-interface.md): bundle
registration and the Mixed Lang
+ task handler interface — `airflow.Bundle`/`Register`/`Serve`; Cross language
TaskFlow: flat positional binding, `arg:` tagged structs, and the untagged
folded-name fallback.
+- [ADR 0008](./adr/0008-native-dag-interface.md): the proposed Native Dag
interface (`airflow.Dag`/
+ `dag.Task`/`airflow.Inputs`/`Before`-`After`)
+
+Cross-cutting Lang-SDK decisions — the coordinator architecture and how
non-Python tasks integrate with
+Airflow core surfaces — are recorded in
[`airflow-core/adr/lang-sdk/`](../airflow-core/adr/lang-sdk).
+Two of them shape the interfaces above:
+[ADR-0008](../airflow-core/adr/lang-sdk/0008-control-flow-constructs.md) for
grouping, conditions,
+branching, and triggering a Dag run, and
+[ADR-0009](../airflow-core/adr/lang-sdk/0009-provider-operators-as-generated-dsl.md)
for reaching
+Python provider operators from a native Dag.
+
+The normative, language-agnostic on-disk bundle format (the footer layout,
manifest fields, and what the
+`ExecutableCoordinator` reads) is specified in
+[`executable-bundle-spec.rst`](https://github.com/apache/airflow/blob/main/task-sdk/docs/executable-bundle-spec.rst).
+`airflow-go-pack` produces bundles conforming to that spec.
+
+## Future Direction
+
+This is more of an "it would be nice to have" than any plan or commitment, and
a place to record ideas.
+
+- The ability to run Airflow tasks "in" an existing code base, i.e. defining
an Airflow task function that
+ runs (in a goroutine) inside an existing app.
+- Doing the task function reflection ahead of time, rather than for each
Execute call.
diff --git a/go-sdk/README.md b/go-sdk/README.md
index 5ca2568faaa..4091be2599b 100644
--- a/go-sdk/README.md
+++ b/go-sdk/README.md
@@ -17,305 +17,93 @@
under the License.
-->
-# Apache Airflow Go Task SDK
+# Airflow Go SDK
-The Go SDK is a Go implementation of the Airflow Task SDK. It lets you write
task functions in Go that
-have native access to the Airflow "model" (Variables, Connections, and XCom),
instead of writing them in
-Python.
+Write Apache Airflow Dags and tasks in Go.
-It is built on the Task Execution Interface (TEI, a.k.a. the Task API)
introduced by AIP-72 in Airflow
-3.0.0. AIP-72 standardised how a task runtime talks to Airflow over an HTTP
Execution API, which decoupled
-the language a task is written in from the Airflow core. The Go SDK is one
such runtime; the Java SDK is
-another.
+Requires Go 1.25 or later.
-> [!WARNING]
-> This is an **experimental** feature. The SDK is under active development and
its APIs, wire protocols,
-> and tooling may change between releases without notice.
+## Installation
-## The compiled-language constraint
-
-Python tasks are imported and run in-process. Go is compiled, so the model is
different.
-
-A single binary that bundles one or more Dags' task functions is called a
**bundle**. You build one with
-the SDK's packer, `airflow-go-pack`, which compiles your code and appends a
metadata footer (the manifest
-of `dag_id`s and `task_id`s, plus the source files) to the executable. The
source files are the
-entrypoint (the file with `func main`) and the file that declares each Dag
built with `airflow.Dag`.
-A Dag declared outside your module, such as in a dependency, uses the
entrypoint as its source. The result is a **self-contained
-executable bundle**: a single runnable file that *is* the bundle, with no
separate manifest or archive to
-ship alongside it.
-
-## You still need a Python stub Dag (for now)
-
-The Task API does not yet carry Dag *structure* for non-Python languages, so
the Dag's shape and task
-dependencies are still declared in a small Python file using `@task.stub`:
-
-```python
-from airflow.sdk import dag, task
-
-
[email protected](queue="golang")
-def extract(): ...
-
-
[email protected](queue="golang")
-def transform(): ...
-
-
-@dag()
-def simple_dag():
- extract() >> transform()
-
-
-simple_dag()
+```bash
+go get github.com/apache/airflow/go-sdk
+go get -tool github.com/apache/airflow/go-sdk/cmd/airflow-go-pack
```
-`@task.stub` tells the Dag parser the "shape" of the Go tasks (their names and
dependencies) without any
-Python implementation. The `queue=` value routes the task to the Go runtime.
This Python requirement is a
-known limitation.
+## Quick start
-
-## Authoring a bundle
-
-Build a bundle with `airflow.Bundle()`, register a handler for each task, and
call `Serve` last. From
-[`example/bundle/main.go`](./example/bundle/main.go):
+Declare a Dag in `main.go`:
```go
-func main() {
- bundle := airflow.Bundle()
-
- bundle.Register(
- airflow.TaskHandler("simple_dag", "extract", extract),
- airflow.TaskHandler("simple_dag", "transform", transform),
- )
-
- if err := bundle.Serve(); err != nil {
- log.Fatal(err)
- }
-}
-```
-
-`TaskHandler` names the `dag_id` and the `task_id` explicitly; neither is
derived from the Go function
-name, so a handler can be called whatever reads best in Go. It checks the
signature of the function and
-panics if the check fails -- a variadic `...` parameter, which no stub
argument can fill, is one such
-failure -- so the executable stops as it starts rather than when the task
first runs. `Serve` then closes
-registration, so a `Register` left below it in `main` panics instead of
changing what the running bundle
-answers for.
-
-A package that defines handlers of its own can export them as a
`[]airflow.Registerable` for `main` to
-pass on with `bundle.Register(reports.Handlers()...)`.
-
-A task is an ordinary Go function whose first parameter is an
[`airflow.Context`](#the-task-context).
-An optional `(any, error)` return becomes the task's XCom; an `error` return
marks the task failed.
+package main
-Every parameter after the Context is a **data parameter**, filled in
declaration order from the
-arguments of the Python stub Task's TaskFlow call. A literal in the Dag file
(`transform("uk", ...)`)
-decodes straight into the parameter; an upstream task's output
(`transform(..., extract())`) is
-pulled from that task's XCom in the current Dag run. If the argument count
doesn't match, or an
-argument's declared type can't fill the Go type, the task fails before its
body runs.
+import (
+ "log"
-Stub parameters the Dag author left at their Python defaults are the
exception: they reach the wire
-but need no Go parameter, so adding a defaulted parameter to a stub doesn't
break the Go functions
-already bound to it.
+ "github.com/apache/airflow/go-sdk/airflow"
+)
-```go
-func extract(actx airflow.Context) (any, error) {
- conn, err := actx.Client().GetConnection(actx, "test_http")
- // ... do work, honour actx cancellation ...
- return map[string]any{"go_version": runtime.Version()}, nil
+func extract(airflow.Context) (int, error) {
+ return 42, nil
}
-// The stub's literal and XCom arguments bind to country and extracted.
-func transform(actx airflow.Context, country string, extracted map[string]any)
error {
- val, err := actx.Client().GetVariable(actx, "my_variable")
- if err != nil {
- return err
- }
- actx.Logger().InfoContext(actx, "Obtained variable", "my_variable", val,
"country", country)
+func report(actx airflow.Context, rows int) error {
+ actx.Logger().InfoContext(actx, "extracted rows", "rows", rows)
return nil
}
-```
-
-The `Register` calls are the single source of truth for which `dag_id`s and
`task_id`s a bundle can run.
-A helper a task calls can still ask for the narrowest interface it needs (e.g.
`sdk.VariableClient`
-instead of `sdk.Client`), which documents the Airflow features it touches and
lets a test pass a fake;
-a handler itself takes the Context and reaches the client through it.
-
-### Name-based struct binding
-
-When a task's **sole data parameter** is a struct, its fields bind **by name**
instead of by
-position — keyword arguments rather than positional ones, and a friendlier
alternative to a long
-flat parameter list. Being the only data parameter is the opt-in; there is no
marker to add.
-
-```go
-type CombineInput struct {
- Region string `arg:"region_code"` // renamed
- Threshold float64
-}
-
-// The Python stub Task calls combine(region_code="uk", threshold=0.5).
-func Combine(actx airflow.Context, input CombineInput) (any, error) {
- return nil, nil
-}
-```
-
-An exported field binds the argument matching its own Go name, folding case
and underscores — so
-`Threshold` takes `threshold` and `RegionCode` would take `region_code`. Reach
for an `arg:"<name>"`
-tag when the names genuinely differ, as `Region` does above. Declaration order
is irrelevant on both
-sides, and embedded structs contribute their fields just as they do to
`encoding/json`.
-
-Taking **more or fewer arguments** than the Python side passes does not fail
the task. The runtime
-logs a warning before the task runs and carries on, one message per direction,
so a call that does
-both at once says so twice:
-
-- `Dag's call passed argument(s) the task handler does not declare`
-- `Task handler declares argument(s) the Dag's call did not pass`
-Name-based struct binding is what keeps a mixed-language task working while
the two sides drift:
-adding a parameter to the stub, or dropping a field from the struct, is a
warning rather than a
-broken Dag. A field nothing matches keeps its Go zero value. Arguments the
call left at their stub
-default are not reported.
-
-A struct that is **not** the sole data parameter is decoded whole from its one
positional argument
-instead, so `arg:` tags only apply to the sole-parameter form; pairing a
tagged struct with other
-data parameters is rejected at registration. A sole struct parameter also
falls back to whole-value
-decoding when it gets exactly one passed argument no field claims, so a task
can still take an
-upstream object as a single argument.
-
-### The task context
-
-`airflow.Context` is the first parameter of every task handler. Everything
Airflow gives the task is a method on it:
-
-| Method | What it returns |
-| --- | --- |
-| `actx.Logger()` | the `*slog.Logger` that writes to the task's Airflow log |
-| `actx.Client()` | the `sdk.Client` for Variables, Connections and XCom |
-| `actx.TaskInstance()` | `DagID`, `RunID`, `TaskID`, `MapIndex` (nil for an
unmapped task), `TryNumber` |
-| `actx.DagRun()` | `DagID`, `RunID`, and the `*time.Time` fields
`LogicalDate`, `DataIntervalStart`, `DataIntervalEnd` (nil when the run has no
such value, e.g. a manual trigger) |
+func main() {
+ dag := airflow.Dag("go_hello", airflow.DagSpec{Schedule: "@daily", Queue:
"golang"})
-It is itself a `context.Context`, so pass it straight to a client call or to
-`http.NewRequestWithContext`, and select on `actx.Done()`, which fires when
the supervisor asks the
-task to stop. Cleanup that must outlive that cancellation runs under
`context.WithoutCancel(actx)`.
-A helper typed as a plain `context.Context` recovers the same surface with
`airflow.FromContext`.
+ extracted := dag.Task(extract)
+ dag.Task(report, airflow.Inputs(extracted))
-```go
-func extract(actx airflow.Context) (any, error) {
- ti := actx.TaskInstance()
- actx.Logger().InfoContext(actx, "running",
- "dag_id", ti.DagID,
- "run_id", ti.RunID,
- "task_id", ti.TaskID,
- "try_number", ti.TryNumber,
- "logical_date", actx.DagRun().LogicalDate,
- )
- return nil, nil
+ bundle := airflow.Bundle()
+ bundle.Register(dag)
+ if err := bundle.Serve(); err != nil {
+ log.Fatal(err)
+ }
}
```
-`airflow.NewContext` builds one, so a handler is an ordinary function call in
a unit test -- see
-[`example/bundle/main_test.go`](./example/bundle/main_test.go):
+Build it into a bundle, a single executable:
-```go
-actx := airflow.NewContext(
- context.Background(), slog.Default(), &mockVars{},
- airflow.TaskInstance{}, airflow.DagRun{},
-)
-err := transform(actx, "uk", map[string]any{"go_version": "go1.24"})
-assert.NoError(t, err)
+```bash
+go tool airflow-go-pack --output dist/go_hello .
```
-### Task logging
-
-In coordinator mode, the logger from `actx.Logger()` filters records using
Airflow's configured `[logging] logging_level` before sending them to the
supervisor. Airflow also propagates `[logging] namespace_levels`; use a
group-scoped logger to set the namespace:
+Put `dist/go_hello` in a Dag bundle, such as the default `dags-folder` bundle,
and send the `golang` queue to the
+Go coordinator in `airflow.cfg`:
-```go
-databaseLog := actx.Logger().WithGroup("example.database")
-databaseLog.Debug("query complete", "rows", 42)
+```ini
+[sdk]
+coordinators = {"go": {"classpath":
"airflow.sdk.coordinators.executable.ExecutableCoordinator"}}
+queue_to_coordinator = {"golang": "go"}
```
-Namespace levels use longest dotted-prefix matching, so an `example=DEBUG`
override also applies to `example.database` unless a more specific override
takes precedence.
-
-## Deployment
-
-A Python task runner executes the Go task directly, with no separate Go worker
process to run on the host.
-This is the same coordinator mechanism the Java SDK uses.
-
-The mature Python supervisor handles the Airflow-facing concerns, so Go tasks
inherit remote task logs,
-the full range of task states, and alternate XCom backends without
implementing those capabilities again.
-
-### Quickstart
-
-- Build and pack your bundle with `airflow-go-pack`. The packer compiles the
bundle and appends an
- embedded metadata footer so the coordinator can read its `dag_id`s without
executing the binary,
- producing a single runnable file:
-
- ```bash
- go tool airflow-go-pack ./example/bundle -- -trimpath -tags=prod
- ```
-
- Use `--output <path>` to write the packed bundle straight into the directory
of the Dag bundle the
- coordinator scans (`task_handler_bundle_name`), and pass extra `go build`
flags after `--`.
+Airflow reads the bundle like any other Dag file: `go_hello` shows up in the
UI, runs on its schedule, and
+`report` receives the value `extract` returned.
- For cross-compiling (e.g. deploy to a Linux host from an Apple-silicon
(darwin/arm64) machine), pass `--goos`/`--goarch` and the
- packer cross-builds for you:
+## Learn more
- ```bash
- go tool airflow-go-pack --goos linux --goarch amd64 \
- --output /opt/airflow/go-task-handlers/sample-dag-bundle \
- ./example/bundle
- ```
+The [Go SDK guide][guide] covers:
- Alternatively, use `--executable`/`--source`. The packer normally execs the
binary to read
- its metadata; a cross-compiled binary cannot run on the host, so generate
the metadata on a machine that
- can run it and pass the file with `--airflow-metadata`:
+| Topic
| Guide section |
+|
---------------------------------------------------------------------------------------------
| ---------------------------------------------- |
+| Task inputs, options, order-only dependencies, task groups, branching,
triggering another Dag | [Declaring a Dag in Go][declaring-a-dag-in-go] |
+| The `airflow.Context`, the `sdk.Client` and the task state store
| [Writing tasks][writing-tasks] |
+| Building a bundle with `airflow-go-pack`, and cross-platform builds
| [Building a bundle][building-a-bundle] |
+| Where to deploy a bundle, and the `ExecutableCoordinator` options
| [Deploying][deploying] |
+| What a Dag declared in Go cannot do yet
| [Limitations][limitations] |
- ```bash
- # on linux/amd64 machine:
- go build -o my-bundle ./example/bundle
- ./my-bundle --airflow-metadata > airflow-metadata.yaml
+See [`example/`](./example) for working projects.
- # on darwin/arm64 machine:
- go tool airflow-go-pack --executable ./my-bundle --source main.go
--airflow-metadata airflow-metadata.yaml
- ```
-
- > [!NOTE]
- > The packer ships via the Go 1.24 `tool` directive, so there is no global
install: add
- > `tool github.com/apache/airflow/go-sdk/cmd/airflow-go-pack` to your bundle
module's `go.mod` and run
- > it with `go tool airflow-go-pack`. This pins the packer version per
project.
-
-- Register a Dag bundle for the packed bundles, register the coordinator, and
route the queue to it, in
- `airflow.cfg` (or the equivalent `AIRFLOW__*` env vars):
-
- ```ini
- [dag_processor]
- dag_bundle_config_list = [
- {"name": "dags-folder", "classpath":
"airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {}},
- {"name": "go-task-handlers", "classpath":
"airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {"path":
"/opt/airflow/go-task-handlers"}}
- ]
-
- [sdk]
- coordinators = {"go": {"classpath":
"airflow.sdk.coordinators.executable.ExecutableCoordinator", "kwargs":
{"task_handler_bundle_name": "go-task-handlers"}}}
- queue_to_coordinator = {"golang": "go"}
- ```
-
- `task_handler_bundle_name` names the Dag bundle the coordinator scans for
packed bundles (the task's own
- Dag bundle when unset). It is used only by mixed-language Dags, to locate
the task handlers for the
- `@task.stub` tasks of a Python Dag; Dags defined natively in a language SDK
do not use it.
- `queue_to_coordinator` routes stub tasks with `queue="golang"` to this Go
coordinator. The matching bundle
- is marked executable before it is launched, so any Dag bundle works,
including an object-store one such as
- `S3DagBundle`.
-
- > [!IMPORTANT]
- > The `[sdk]` config and the packed bundle files must be present wherever
tasks execute and on the Dag
- > processor. With `CeleryExecutor`, tasks execute on the Celery workers;
with `LocalExecutor`, they run
- > inside the scheduler process. The Dag processor checks the stub tasks of
each Python Dag against the
- > task handlers the packed bundles register, so it runs them too and needs
bundles built for its
- > operating system and CPU architecture. The API server does not need any of
it. Register the Dag bundle
- > in `[dag_processor] dag_bundle_config_list` on every component, like your
other Dag bundles: the worker
- > and the Dag processor resolve `task_handler_bundle_name` through it, and
wherever the `[sdk]` config is
- > read it is rejected if the name is missing there.
-
-- Deploy the matching Python stub Dag (above) into Airflow. There is no
separate Go worker to run: the
- Airflow worker forks the bundle binary once per task instance.
+[guide]:
https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/language-sdks/go.html
+[declaring-a-dag-in-go]:
https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/language-sdks/go.html#declaring-a-dag-in-go
+[writing-tasks]:
https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/language-sdks/go.html#writing-tasks
+[building-a-bundle]:
https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/language-sdks/go.html#building-a-bundle
+[deploying]:
https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/language-sdks/go.html#deploying
+[limitations]:
https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/language-sdks/go.html#limitations
## Compatibility matrix
@@ -372,185 +160,10 @@ prek hook regenerate it.
<!-- END AUTO-GENERATED LANG-SDK COMPAT MATRIX -->
-## How it works
-
-The bundle binary speaks the coordinator protocol when Airflow launches it
with `--comm` and `--logs`.
-
-```
-Python supervisor / task runner
- │ finds + validates the bundle, then forks it:
- ▼
- <bundle binary> --comm=127.0.0.1:P1 --logs=127.0.0.1:P2
- │ binary dials BACK over TCP loopback (msgpack-over-IPC)
- ▼
- GetConnection / GetVariable / GetXCom / SetXCom ... → SucceedTask / TaskState
-```
-
-- The Python `ExecutableCoordinator` forks the bundle binary with
`--comm`/`--logs` addresses it is already
- listening on. The binary dials back (it never listens) and speaks a
length-prefixed msgpack-over-IPC wire
- protocol on the comm socket, with structured JSON-line logs on the logs
socket.
-- The Python runtime is the worker. It proxies every `GetConnection` /
`GetVariable` / `GetXCom` /
- `SetXCom` call through to the Execution API. The Go binary just runs the
task function.
-- When the first frame on the comm socket is a `DagFileParseRequest` from the
Dag processor, the
- binary runs no task. It answers with one `DagFileParsingResult` that holds
the Dags from
- `airflow.Dag` that the binary registered. The Dags are serialized as
- [Serializing a native Dag](#serializing-a-native-dag) describes. The binary
then exits.
-
-The Go side of the protocol is implemented in `pkg/execution/`. On the Python
side it is the
-`ExecutableCoordinator` in
`task-sdk/src/airflow/sdk/coordinators/executable/coordinator.py`.
-
-## The vendored schemas
-
-`schema/dag-schema.json` and `schema/supervisor-schema.json` are byte-for-byte
copies of schemas
-airflow-core and task-sdk own. The generators read the copies, not the
originals, so a generated file
-is explainable from a file inside this module and a standalone checkout — a
published Go module, an
-ASF source release — can regenerate and test without the monorepo around it.
`ts-sdk` and `java-sdk`
-vendor theirs the same way.
-
-Two prek hooks split the work that keeps the copies honest.
`sync-go-sdk-schemas` copies an original
-over its copy when the two differ and fails, since copying is mechanical; it
is a **manual** hook, so
-re-vendoring is a deliberate step here rather than something a change to a
Python schema forces on
-whoever made it — which also means nothing announces on its own that a copy
has gone stale.
-`check-go-sdk-generated-drift` runs on every commit and regenerates from the
copies, failing when the
-committed Go differs, since what to do about a new schema construct — a
generator rule, an authoring
-exclusion — is a decision.
-
-## Regenerating the coordinator-protocol models
-
-The types in `pkg/execution/genmodels/` are generated from
`schema/supervisor-schema.json`, this
-module's vendored copy of the supervisor schema snapshot the Python Task SDK
owns
-(`task-sdk/src/airflow/sdk/execution_time/schema/schema.json`); do not edit
either by hand. To move
-the SDK to a newer schema version:
-
-1. Refresh the copy: `prek run sync-go-sdk-schemas --hook-stage manual`, which
overwrites
- `schema/supervisor-schema.json` from task-sdk's and fails so the change
lands in review.
-2. Set `SupervisorSchemaVersion` in
[`pkg/execution/messages.go`](./pkg/execution/messages.go) to the
- copy's `api_version` date.
-3. Run `just generate-models`, and commit the copy with what it generated.
-
-`TestSupervisorSchemaVersionMatchesSnapshot` fails when the constant and the
copy's `api_version`
-drift, so a missed bump is caught by `go test`. A snapshot can also grow a
field without the
-`api_version` moving, which leaves the models behind with nothing failing —
msgpack drops a field the
-struct does not declare — so the `check-go-sdk-generated-drift` prek hook
regenerates them and fails
-when the committed files differ.
-
-[`airflow/enums.go`](./airflow/enums.go) declares the `genmodels.DagRunState`
constants again, so that
-a Dag author does not import `genmodels`. `TestDagRunStateMatchesGenmodels`
fails when regenerating
-the models adds, renames or removes a `DagRunState` constant. The test keeps
failing until
-`airflow/enums.go` declares the same constants as `genmodels`.
-
-## Regenerating the Dag, task and task group specs
-
-`airflow.DagSpec`, `airflow.TaskSpec` and `airflow.TaskGroupSpec` in
-[`airflow/spec.gen.go`](./airflow/spec.gen.go) are generated from
`schema/dag-schema.json`, this
-module's vendored copy of airflow-core's Dag serialization schema
-(`airflow-core/src/airflow/serialization/schema.json`), which Python owns; do
not edit any of them
-by hand. Refresh the copy with `prek run sync-go-sdk-schemas --hook-stage
manual`,
-then run `just generate-specs` after changing the schema or the generator.
-
-The schema is the serialized shape rather than the authoring one, so
-[`internal/genspec/authoring.go`](./internal/genspec/authoring.go) holds the
three tables that turn
-it into the authoring shape, each entry carrying the reason it exists:
-
-- **exclusions** — properties an author never sets, such as the paths the
bundle fills in and the
- template fields of a Python operator class. A property *not* excluded
generates, so one added on
- the Python side surfaces in review instead of vanishing.
-- **type overrides** — the schema types a moment in time and a duration as a
number of seconds, and
- an integral count as a JSON number.
-- **injections** — `Schedule`, which stands in for the serialized `timetable`,
and the `Queue` of
- `DagSpec`, for which the schema has no Dag-level property. Injecting into
the schema rather than
- hand-writing the field keeps every field in one struct declaration, which is
what lets
- `TaskSpec` implement the sealed `TaskOption`.
-
-The schema types `trigger_rule` and `weight_rule` as plain strings and does
not list their values.
-`TriggerRule`, `WeightRule` and their constants are therefore hand-written in
-[`airflow/enums.go`](./airflow/enums.go). `TestRuleConstantsMatchPython`
checks those constants
-against the Python enums in airflow-core.
-
-`just generate-specs` also writes
[`airflow/spec_fields.gen.go`](./airflow/spec_fields.gen.go), which
-holds the schema key and the schema default of each field, keyed by the name
of the Go field.
-
-The `check-go-sdk-generated-drift` prek hook regenerates both files and fails
when a committed one
-differs, so a schema change that never reached Go cannot merge.
-
-## Serializing a native Dag
-
-[`airflow/serialize.go`](./airflow/serialize.go) writes a registered
`airflow.Dag` as the Dag JSON
-that Airflow stores. Python's `DagSerialization` writes the same shape for a
Python Dag. Each field
-of `DagSpec`, `TaskSpec` and `TaskGroupSpec` has an entry in a generated
table, which names the
-schema property that the field sets and its schema default. The serializer
leaves out a value equal to
-the default. `serialize.go` lists only the fields that the generated tables
cannot describe: the ones
-it writes in its own way, and the lists that Python keeps in a set.
-
-`airflow.Inputs` takes the results of tasks and `airflow.Literal` values, one
for each parameter after
-the `Context`. A task becomes an `xcom` argument binding and a literal becomes
a `literal` one:
-
-```go
-dag.Task(load, airflow.Inputs(transformed, airflow.Literal("s3://bucket/out")))
-```
-
-```json
-[
- {"name": "arg0", "kind": "xcom", "task_id": "transform"},
- {"name": "arg1", "kind": "literal", "value": "s3://bucket/out"}
-]
-```
-
-A literal must be JSON that decodes into its parameter type, and it adds no
edge to the Dag.
-
-The `check-go-sdk-serialization-conformance` prek hook builds the Dags of
-[`scripts/ci/lang_sdk_serialization/test_dags.yaml`](../scripts/ci/lang_sdk_serialization/test_dags.yaml)
-twice: with this SDK, and with Python and Airflow's own serializer. It loads
the Go output with
-Airflow's deserializer and compares the two serializations field by field. Run
it from the
-repository root:
-
-```bash
-prek run check-go-sdk-serialization-conformance --all-files
-```
+## Links
-The hook runs every Dag of the file, including the ones that `requires` a
feature that only some SDKs
-have.
[`go_native.json`](../airflow-core/tests/unit/dag_processing/lang_sdk_fixtures/go_native.json)
-holds the Go output for the same Dags, so that the Airflow core tests check
that Airflow loads it.
-`TestSerializeConformanceDags` says how to rewrite it.
-
-## Architectural decisions
-
-The [`adr/`](./adr) directory records the design decisions behind the SDK:
-
-- [ADR 0001](./adr/0001-bundle-packing-options.md): bundle-packing options.
-- [ADR 0002](./adr/0002-use-go-tool-directive-for-bundle-packer.md): deliver
the bundle packer via the
- Go 1.24 `tool` directive.
-- [ADR 0003](./adr/0003-coordinator-protocol-msgpack-ipc.md): the
msgpack-over-IPC coordinator protocol.
-- [ADR 0004](./adr/0004-self-contained-executable-bundle.md): the
self-contained executable bundle, where
- the executable *is* the bundle.
-- [ADR 0005](./adr/0005-retire-go-edge-worker.md): retire the standalone Go
Edge Worker and make the
- coordinator the only execution path.
-- [ADR 0006](./adr/0006-cross-language-argument-binding.md): cross-language
TaskFlow argument binding,
- from serialized literals and XCom references to typed Go parameters.
-- [ADR 0007](./adr/0007-mixed-lang-task-handler-interface.md): bundle
registration and the Mixed Lang
- task handler interface — `airflow.Bundle`/`Register`/`Serve`; Cross language
TaskFlow: flat positional binding, `arg:` tagged structs, and the untagged
folded-name fallback.
-- [ADR 0008](./adr/0008-native-dag-interface.md): the proposed Native Dag
interface (`airflow.Dag`/
- `dag.Task`/`airflow.Inputs`/`Before`-`After`)
-
-Cross-cutting Lang-SDK decisions — the coordinator architecture and how
non-Python tasks integrate with
-Airflow core surfaces — are recorded in
[`airflow-core/adr/lang-sdk/`](../airflow-core/adr/lang-sdk).
-Two of them shape the interfaces above:
-[ADR-0008](../airflow-core/adr/lang-sdk/0008-control-flow-constructs.md) for
grouping, conditions,
-branching, and triggering a Dag run, and
-[ADR-0009](../airflow-core/adr/lang-sdk/0009-provider-operators-as-generated-dsl.md)
for reaching
-Python provider operators from a native Dag.
-
-The normative, language-agnostic on-disk bundle format (the footer layout,
manifest fields, and what the
-`ExecutableCoordinator` reads) is specified in
-[`executable-bundle-spec.rst`](https://github.com/apache/airflow/blob/main/task-sdk/docs/executable-bundle-spec.rst).
-`airflow-go-pack` produces bundles conforming to that spec.
-
-## Future Direction
-
-This is more of an "it would be nice to have" than any plan or commitment, and
a place to record ideas.
-
-- Defining the whole Dag in the Go SDK, so the Python stub Dag is no longer
required and a bundle's
- structure and task dependencies can be declared natively in Go.
-- The ability to run Airflow tasks "in" an existing code base, i.e. defining
an Airflow task function that
- runs (in a goroutine) inside an existing app.
-- Doing the task function reflection ahead of time, rather than for each
Execute call.
+- [Go SDK guide][guide]
+- [API reference](https://pkg.go.dev/github.com/apache/airflow/go-sdk) on
pkg.go.dev
+- [Source](https://github.com/apache/airflow/tree/main/go-sdk): the `go-sdk/`
directory of the Apache Airflow monorepo
+- [Issues](https://github.com/apache/airflow/issues): bug reports and feature
requests
+- [Developing this module](./DEVELOPMENT.md): building, testing, code
generation and design decisions
diff --git a/go-sdk/example/README.md b/go-sdk/example/README.md
new file mode 100644
index 00000000000..bcaea1bc8f3
--- /dev/null
+++ b/go-sdk/example/README.md
@@ -0,0 +1,129 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+ -->
+
+# Go SDK examples
+
+Example Dags for the Apache Airflow Go SDK. See the [Go SDK guide][guide] for
the full documentation.
+
+[guide]:
https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/language-sdks/go.html
+
+## Build
+
+Run these from `go-sdk/`. Each command writes one bundle into `bin/`:
+
+```bash
+# Dag definition
+go tool airflow-go-pack --output bin/go_native ./example/native
+
+go build -o bin/dynamic/go_dynamic_dags.raw ./example/dynamic
+cp example/dynamic/dags.yaml bin/dynamic/dags.yaml
+go tool airflow-go-pack --executable bin/dynamic/go_dynamic_dags.raw \
+ --source example/dynamic/main.go --output bin/go_dynamic_dags
+
+# Python Dag with stub TaskHandler
+go tool airflow-go-pack --output bin/example_dags ./example/bundle
+```
+
+The dynamic example reads `dags.yaml` next to its binary, including when
`airflow-go-pack` reads its Dags, so it
+is built first and packed in place.
+
+## Configure Airflow
+
+Lay out the bundles like this:
+
+```text
+<dags_folder>/
+ go/go_native
+ go-dynamic/go_dynamic_dags
+ go-dynamic/dags.yaml
+ go_examples.py # copied from go-sdk/dags/
+/opt/airflow/go-task-handlers/
+ example_dags
+ .airflowignore # containing "*"
+```
+
+Register `go-task-handlers` as a Dag bundle, and route the `golang` queue to
the Go coordinator:
+
+```bash
+export AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST='[
+ {"name": "dags-folder", "classpath":
"airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {}},
+ {
+ "name": "go-task-handlers",
+ "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
+ "kwargs": {"path": "/opt/airflow/go-task-handlers"}
+ }
+]'
+export AIRFLOW__SDK__COORDINATORS='{
+ "go": {
+ "classpath": "airflow.sdk.coordinators.executable.ExecutableCoordinator",
+ "kwargs": {"task_handler_bundle_name": "go-task-handlers"}
+ }
+}'
+export AIRFLOW__SDK__QUEUE_TO_COORDINATOR='{"golang": "go"}'
+```
+
+The bundles must be built for the platform of the workers and the Dag
processor.
+
+The examples read this Connection:
+
+```bash
+airflow connections add test_http --conn-type http --conn-host example.com
+```
+
+## Dag definition
+
+These Dags are declared entirely in Go:
+
+| Dag | Source
| Shows
|
+| ---------------------------------------------- | --------------------------
|
--------------------------------------------------------------------------------------
|
+| `go_native_pipeline` | `native/main.go`
| Task inputs, literals, a task group, an edge label, `If`, `Switch`, and a
trigger rule |
+| `go_native_trigger` | `native/main.go`
| `TriggerDagRun`, which starts `go_native_report`
|
+| `go_native_report` | `native/report.go`
| A Dag declared in its own file, which the Code view shows
|
+| `go_dynamic_a`, `go_dynamic_b`, `go_dynamic_c` | `dynamic/` and `dags.yaml`
| One Dag per entry of `dags.yaml`, which changes the Dags with no rebuild
|
+
+`go_native_pipeline` reads two Variables. Set `go_native_cadence` to `weekly`
to take the other branch of
+`Switch`:
+
+```bash
+airflow variables set go_native_greeting "hello from Airflow"
+airflow variables set go_native_cadence daily
+
+airflow dags trigger go_native_pipeline
+airflow dags trigger go_native_trigger
+```
+
+## Python Dag with stub TaskHandler
+
+The Python Dags in `go-sdk/dags/go_examples.py` declare stub tasks, and
`bundle/` implements them:
+
+| Dag | Shows
|
+| ---------------------- |
------------------------------------------------------------------------------ |
+| `simple_dag` | Python and Go tasks in one Dag, TaskFlow arguments,
Connections, and a retry |
+| `taskflow_binding_dag` | Each way a stub task's arguments bind to a Go
handler: by position and by name |
+| `concurrent_xcom_dag` | Reading XComs from goroutines
|
+| `variable_write_dag` | Writing and deleting Variables
|
+| `task_state_dag` | The task state store
|
+
+`simple_dag` reads one Variable:
+
+```bash
+airflow variables set my_variable "hello from Airflow"
+
+airflow dags trigger simple_dag
+```
