> ## Documentation Index > Fetch the complete documentation index at: https://astronomer.io/docs/llms.txt > Use this file to discover all available pages before exploring further. # Run Airflow tasks in other languages Airflow 3 enables users to write SDKs allowing definition of Airflow tasks in languages other than Python. Experimental SDKs for Golang and Java are available as of the Task SDK 1.3 release. Support for other languages: * Makes it easier for users to migrate workflows from legacy tools written in languages other than Python to Airflow. * Makes Airflow more accessible to developers who prefer to code in another language. * Gives users access to features unique to a supported language. Multilanguage support is currently experimental and under development. This guide is subject to change and will be expanded over time. If you want to contribute to support writing Airflow tasks in the language of your choice, contact the Airflow developers in the [Airflow Slack](https://apache-airflow-slack.herokuapp.com/) or the [Airflow Dev list](https://airflow.apache.org/community/). ## Assumed knowledge To get the most out of this guide, you should have existing knowledge of: * Basic Airflow concepts. See [Introduction to Apache Airflow](/docs/learn/intro-to-airflow). * Depending on your target language: * Basic Golang concepts. See [Golang documentation](https://go.dev/doc/). * Basic Java concepts. See [Learn Java](https://dev.java/learn/). ## How it works Two environment variables configure which SDK coordinators are available, in addition to the default Python Task SDK: * `AIRFLOW__SDK__COORDINATORS`: a JSON containing all available coordinators. * `AIRFLOW__SDK__QUEUE_TO_COORDINATOR`: a JSON that maps `queue` names to coordinator names. When adding tasks in other languages, you write the task logic in your target language in a separate module and if needed, compile it. Two elements need to be present: the `dag_id`, matching the id of the Dag in which the task is used, and a `task_id` that matches the id of the function decorated with `@task.stub` in the Dag. The Dag itself is written in Python. ```python wrap theme={null} from airflow.sdk import task @task.stub(queue="my-language-queue") def my_task(): ... my_task() ``` When the task runs, Airflow uses the queue, `dag_id`, and `task_id` to find the matching function in the target language and executes it. How you declare the ids differs by language, as shown in the examples below. Additionally, the SDKs contain a client for reading Variables, Connections, and XCom, as well as a logger. A value returned from the task becomes its XCom. The compiled bundle and the coordinator configuration must be available on the Airflow component that runs your tasks. When using task SDKs for other languages on Astro, you need to create matching [worker queues](/docs/astro/configure-worker-queues) in addition to setting `AIRFLOW__SDK__COORDINATORS` and `AIRFLOW__SDK__QUEUE_TO_COORDINATOR` as [environment variables](/docs/astro/manage-env-vars). For example, if you are using the Golang SDK, and set `AIRFLOW__SDK__QUEUE_TO_COORDINATOR='{"golang": "go"}'`, you need to create a worker queue with the name `golang`. Additionally, make sure any compiled binaries (for example, Go executables) are compatible with `linux/amd64`, the architecture of workers on Astro Hosted. ## Golang SDK example The Golang SDK is experimental and still under development. You can track its status [here](https://pkg.go.dev/github.com/apache/airflow/go-sdk). Make sure your Airflow project is at least on version 3.3 and using the Task SDK version 1.3+. ### Step 1: Configure the executable coordinator Add two environment variables to your `.env` file. The first maps the `golang` queue to a coordinator named `go`. The second defines that coordinator, which scans the `executables_root` location for compiled bundles and runs them. The executables location needs to be accessible to your Airflow worker. ```text wrap theme={null} AIRFLOW__SDK__QUEUE_TO_COORDINATOR='{"golang": "go"}' AIRFLOW__SDK__COORDINATORS='{ "go": { "classpath": "airflow.sdk.coordinators.executable.ExecutableCoordinator", "kwargs": { "executables_root": ["/usr/local/airflow/include/go_bundle/bin"] } } }' ``` ### Step 2: Write a Go task bundle 1. Create the directory for the Go module, the parent directory of the `executables_root`: ```sh wrap theme={null} $ mkdir -p include/go_bundle && cd include/go_bundle ``` 2. Create a `go.mod` file. The `require` line pins the SDK version, and the `tool` directive makes the bundle packer available through `go tool`. Replace the placeholders with your versions. ```text wrap theme={null} module example.com/go-bundle go require github.com/apache/airflow/go-sdk tool github.com/apache/airflow/go-sdk/cmd/airflow-go-pack ``` 3. Create a `main.go` file with the task logic. ```go expandable wrap theme={null} package main import ( "encoding/json" "fmt" "log" "log/slog" "runtime" v1 "github.com/apache/airflow/go-sdk/bundle/bundlev1" "github.com/apache/airflow/go-sdk/bundle/bundlev1/bundlev1server" "github.com/apache/airflow/go-sdk/sdk" ) var ( bundleName = "go_task_syntax_example" bundleVersion = "1.0.0" ) type bundle struct{} var _ v1.BundleProvider = (*bundle)(nil) func (b *bundle) GetBundleVersion() v1.BundleInfo { return v1.BundleInfo{Name: bundleName, Version: &bundleVersion} } func (b *bundle) RegisterDags(dagbag v1.Registry) error { d := dagbag.AddDag("go_task_syntax_example") d.AddTask(transform) return nil } func main() { if err := bundlev1server.Serve(&bundle{}); err != nil { log.Fatal(err) } } func transform(ctx sdk.TIRunContext, client sdk.Client, logger *slog.Logger) (any, error) { ti := ctx.TaskInstance() raw, err := client.GetXCom(ctx, ti.DagID, ti.RunID, "extract", nil, "return_value", nil) if err != nil { return nil, fmt.Errorf("reading extract XCom: %w", err) } var payload struct { Numbers []float64 `json:"numbers"` } encoded, err := json.Marshal(raw) if err != nil { return nil, fmt.Errorf("re-encoding extract payload: %w", err) } if err := json.Unmarshal(encoded, &payload); err != nil { return nil, fmt.Errorf("decoding extract payload: %w", err) } var sum float64 for _, n := range payload.Numbers { sum += n } logger.Info("summed numbers in Go", "sum", sum, "count", len(payload.Numbers)) return map[string]any{ "sum": sum, "count": len(payload.Numbers), "computed_by": "Go " + runtime.Version(), }, nil } ``` The `RegisterDags` method binds Go functions to the Python Dag. The `dag_id` you pass to `AddDag` must match the Python `@dag` id, and each function name passed to `AddTask` must match a Python stub task name. The `transform` function reads the `extract` task's XCom, sums the numbers, and pushes the result to XCom. ### Step 3: Build the bundle From the `include/go_bundle` directory, compile the bundle for the architecture of your Airflow containers (`--goos linux` for Linux containers). Use `arm64` on Apple Silicon or `amd64` on Intel and AMD machines. Note that you need [Go](https://go.dev/dl/) 1.24 or later to compile the task bundle. ```sh wrap theme={null} $ go mod tidy $ go tool airflow-go-pack --goos linux --goarch arm64 --output ./bin/go_task_syntax_example . ``` This writes a single executable to `include/go_bundle/bin`, which is the `executables_root` you set in Step 1. Astro Hosted workers use `linux/amd64`, which means you'll need to compile with `--goos linux --goarch amd64` before deploying your project to Astro. ### Step 4: Create the Dag In your `dags` folder, create a file called `go_task_syntax_example.py` with the following code: ```python wrap theme={null} import random from airflow.sdk import dag, task, chain @dag(tags=["go sdk"]) def go_task_syntax_example(): @task def extract(): return {"numbers": [random.randint(1, 100) for _ in range(random.randint(3, 6))]} @task.stub(queue="golang") def transform(): ... @task def load(result): print(f"Go returned {result}") return result extracted = extract() transformed = transform() chain(extracted, transformed) load(transformed) go_task_syntax_example() ``` The Python `extract` task pushes a list of numbers to XCom, the Go `transform` task reads that list and sums it, and the Python `load` task reads the result back from Go. The `transform` task uses `@task.stub(queue="golang")` and has no Python body. The stub tells Airflow the task's name and its place in the Dag, and the `queue` value routes it to the Go coordinator. The `dag_id` and the stub task name must match the values registered in the Go bundle. When using the Golang SDK on Astro, you need to create a matching [worker queue](/docs/astro/configure-worker-queues) in addition to setting `AIRFLOW__SDK__COORDINATORS` and `AIRFLOW__SDK__QUEUE_TO_COORDINATOR` as [environment variables](/docs/astro/manage-env-vars). For example, for `AIRFLOW__SDK__QUEUE_TO_COORDINATOR='{"golang": "go"}'`, you need to create a worker queue with the name `golang`. ## Java SDK example The Java SDK is experimental and still under development. You can track its status in the [java-sdk directory](https://github.com/apache/airflow/tree/main/java-sdk) of the Airflow repository. Make sure your Airflow project is at least on version 3.3 and using the Task SDK version 1.3+. The Java task runs as a compiled jar, so the Airflow component that runs your tasks also needs a Java runtime, for example `openjdk-21-jre-headless`. You'll need to add the runtime to your `packages.txt` file to install it in your image. ### Step 1: Configure the Java coordinator Add two environment variables to your `.env` file. The first maps the `java` queue to a coordinator named `java`. The second defines that coordinator, which scans the `jars_root` location for compiled bundle jars and runs them. ```text wrap theme={null} AIRFLOW__SDK__QUEUE_TO_COORDINATOR='{"java": "java"}' AIRFLOW__SDK__COORDINATORS='{ "java": { "classpath": "airflow.sdk.coordinators.java.JavaCoordinator", "kwargs": { "jars_root": ["/usr/local/airflow/include/java_bundle"] } } }' ``` ### Step 2: Write a Java task bundle 1. Create the module directory, including the `com/example/bundle` package path where the source files live. You run Gradle from the module root, so change into `include/java_sdk`: ```sh wrap theme={null} $ mkdir -p include/java_sdk/src/java/com/example/bundle $ cd include/java_sdk ``` 2. Create the Gradle project files. `gradle.properties` sets the SDK version in one place, `settings.gradle` names the project and points Gradle at the Apache snapshot repository, and `build.gradle` applies the Airflow SDK plugin, pulls in the SDK and its annotation processor, and points `airflowBundle` at the bundle's main class. Set `projectVersion` to the SDK version you are targeting; current builds are published as snapshots. `gradle.properties`: ```text wrap theme={null} org.gradle.configuration-cache=true projectVersion=1.0.0-SNAPSHOT ``` `settings.gradle`: ```text wrap theme={null} pluginManagement { repositories { maven { url "https://repository.apache.org/content/repositories/snapshots/" mavenContent { snapshotsOnly() } } gradlePluginPortal() mavenCentral() } } rootProject.name = "airflow-java-sdk-etl-example" ``` `build.gradle`: ```text expandable wrap theme={null} plugins { id("org.apache.airflow.sdk") version "${projectVersion}" } repositories { maven { url "https://repository.apache.org/content/repositories/snapshots/" mavenContent { snapshotsOnly() } } mavenCentral() } dependencies { annotationProcessor("org.apache.airflow:airflow-sdk-processor:${projectVersion}") implementation("org.apache.airflow:airflow-sdk:${projectVersion}") implementation("org.slf4j:slf4j-simple:2.0.17") } java { sourceCompatibility = JavaVersion.VERSION_17 targetCompatibility = JavaVersion.VERSION_17 } sourceSets { main { java.srcDir("src/java") resources.srcDir("src/resources") } } airflowBundle { mainClass = "com.example.bundle.EtlBundleBuilder" } ``` 3. Create the task class at `src/java/com/example/bundle/JavaEtlExample.java`. The `@Builder.Dag` and `@Builder.Task` annotations set the ids, and `@Builder.XCom(task = "extract")` adds the upstream Python task's XCom as a method parameter. The returned `Map` becomes the task's XCom. ```java expandable wrap theme={null} package com.example.bundle; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import org.apache.airflow.sdk.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @Builder.Dag(id = "java_task_syntax_example") public class JavaEtlExample { private static final Logger logger = LoggerFactory.getLogger(JavaEtlExample.class); @Builder.Task(id = "transform") public Map transform( Client client, @Builder.XCom(task = "extract") Map payload) { logger.info("[transform/java] received payload from python 'extract' task: {}", payload); List numbers = (List) payload.get("numbers"); long sum = 0; for (Object n : numbers) { sum += ((Number) n).longValue(); } Map result = new LinkedHashMap<>(); result.put("sum", sum); result.put("count", numbers.size()); result.put("computed_by", "Java " + System.getProperty("java.version")); logger.info("[transform/java] summed {} numbers to {}", numbers.size(), sum); return result; } } ``` 4. Create the bundle entry point at `src/java/com/example/bundle/EtlBundleBuilder.java`. It implements `BundleBuilder`, registers the Dag classes, and serves the bundle from `main`. ```java wrap theme={null} package com.example.bundle; import java.util.List; import org.apache.airflow.sdk.*; public class EtlBundleBuilder implements BundleBuilder { @Override public Iterable getDags() { return List.of(JavaEtlExampleBuilder.build()); } public static void main(String[] args) { var bundle = new EtlBundleBuilder().build(); Server.create(args).serve(bundle); } } ``` `JavaEtlExampleBuilder` is generated at compile time by the annotation processor from the `@Builder` annotations on `JavaEtlExample`. ### Step 3: Build the bundle From the `include/java_sdk` directory, build the bundle jar with Gradle, then copy the jar into the `jars_root` you set in Step 1. Building needs a JDK (this example builds with Java 21). ```sh wrap theme={null} $ gradle bundle $ cp build/bundle/*.jar ../java_bundle/ ``` The coordinator loads the jar from `include/java_bundle`. ### Step 4: Create the Dag In your `dags` folder, create a file called `java_task_syntax_example.py` with the following code: ```python wrap theme={null} import random from airflow.sdk import dag, task, chain @dag(tags=["Java SDK"]) def java_task_syntax_example(): @task def extract(): return {"numbers": [random.randint(1, 100) for _ in range(random.randint(3, 6))]} @task.stub(queue="java") def transform(): ... @task def load(result): print(f"Java returned {result}") return result extracted = extract() transformed = transform() chain(extracted, transformed) load(transformed) java_task_syntax_example() ``` The `transform` task uses `@task.stub(queue="java")` and has no Python body. The `queue` value routes it to the Java coordinator, and the `dag_id` and stub task name must match the values registered in the Java bundle. When using the Java SDK on Astro, you need to create a matching [worker queue](/docs/astro/configure-worker-queues) in addition to setting `AIRFLOW__SDK__COORDINATORS` and `AIRFLOW__SDK__QUEUE_TO_COORDINATOR` as [environment variables](/docs/astro/manage-env-vars). For example, for `AIRFLOW__SDK__QUEUE_TO_COORDINATOR='{"java": "java"}'`, you need to create a worker queue with the name `java`. ## Other ways to run tasks in other languages You can also run tasks in other languages using the following methods: * Use the `BashOperator` to run a script in another language. For example, you can use the `BashOperator` to run a JavaScript or R script. See [Run a script in another programming language](/docs/learn/bashoperator#example-run-a-script-in-another-programming-language) for more information. * Use the `KubernetesPodOperator` to run any Docker image, which can include code in any language. See [Use the `KubernetesPodOperator` to run a script in another language](/docs/learn/kubepod-operator#example-use-the-kubernetespodoperator-to-run-a-script-in-another-language) for more information.