jason810496 commented on code in PR #74698:
URL: https://github.com/apache/airflow/pull/74698#discussion_r4345363605
##########
airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst:
##########
@@ -548,123 +628,154 @@ 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 the Dag processor
needs the same ``[sdk]``
+configuration as the workers, and a bundle built for its platform:
-Parsing native Dags
-~~~~~~~~~~~~~~~~~~~
+.. code-block:: ini
+
+ [sdk]
+ coordinators = {"go": {"classpath":
"airflow.sdk.coordinators.executable.ExecutableCoordinator"}}
+ queue_to_coordinator = {"golang": "go"}
-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>`.
+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.
-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:
+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.
+
+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 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.
+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).
+
+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 the scheduler's
+process, so set them there. The Dag processor needs them too, as described
above. Register every Dag bundle in
Review Comment:
Fixed the same way as the Java PR: with `LocalExecutor`, tasks run in
subprocesses on the scheduler's host.
##########
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"}
Review Comment:
Unified: the quick start keeps the only copy, and Deploying links to it. The
third block sets `task_handler_bundle_name`, so it stays.
##########
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")))
- 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``.
+``publish`` receives the ``Summary`` that ``summarize`` returned, and the
literal ``"nightly"`` as ``label``.
-Go entry point
-~~~~~~~~~~~~~~~
+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.
-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.
+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
+~~~~~~~~
+
+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: "..."}``.
+
+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.
+
+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 side not taken is skipped, and stays skipped if it is cleared later. Only
that task is skipped. A task after both
+sides follows its own trigger rule, and the default ``all_success`` skips it
as well. To run such a task when at least
+one side succeeds and none fails, give it
``airflow.TriggerRuleNoneFailedMinOneSuccess``:
Review Comment:
Took your wording.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
