FrankYang0529 commented on code in PR #74698: URL: https://github.com/apache/airflow/pull/74698#discussion_r4310333368
########## airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst: ########## @@ -22,268 +22,458 @@ Go SDK |experimental| -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. - -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. +The Go SDK lets you write Airflow Dags and tasks in Go. There are two ways to use it: + +* **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`. + +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. It is in **beta**, and its API may change. + +.. 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. - 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``. +Task ids +~~~~~~~~ -Go entry point -~~~~~~~~~~~~~~~ +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: "..."}``. -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. +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`` to ``false`` in an ``airflow.TaskGroupSpec`` 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 the tasks named here are skipped, +so a task that both sides lead to still runs. This differs from Python's ``@task.branch``, which skips every +direct downstream task it did not choose. Review Comment: This paragraph says that "a task that both sides lead to still runs". With the default trigger rule, Airflow skips that task. In go doc, it mentions "A task that runs after both the task from Then and the task from Else needs a trigger rule that runs it when one of the two is skipped, such as [TriggerRuleNoneFailedMinOneSuccess]." https://github.com/apache/airflow/blob/777bf9b18881f59a9fb2bfd55f033dbe8904d15c/go-sdk/airflow/if.go#L73-L75 How about changing this paragraph like following: ```suggestion 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``: .. code-block:: go dag.Task(finish, airflow.TaskSpec{ TriggerRule: airflow.TriggerRuleNoneFailedMinOneSuccess, }).After(loaded, reportedEmpty) ``` -- 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]
