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.

Assumed knowledge

To get the most out of this guide, you should have existing knowledge of:

• Basic Airflow concepts. See Introduction to Apache Airflow.
• Depending on your target language:
  • Basic Golang concepts. See Golang documentation.
  • Basic Java concepts. See Learn Java.

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.

1
from airflow.sdk import task
2
3
@task.stub(queue="my-language-queue")
4
def my_task(): ...
5
6
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.

Golang SDK example

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.

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:

   1
   $ 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.

   module example.com/go-bundle
   go <your-go-version>
   require github.com/apache/airflow/go-sdk <your-go-sdk-version>
   tool github.com/apache/airflow/go-sdk/cmd/airflow-go-pack

3. Create a main.go file with the task logic.

   1
   package main
   2
   3
   import (
   4
       "encoding/json"
   5
       "fmt"
   6
       "log"
   7
       "log/slog"
   8
       "runtime"
   9
   10
       v1 "github.com/apache/airflow/go-sdk/bundle/bundlev1"
   11
       "github.com/apache/airflow/go-sdk/bundle/bundlev1/bundlev1server"
   12
       "github.com/apache/airflow/go-sdk/sdk"
   13
   )
   14
   15
   var (
   16
       bundleName    = "go_task_syntax_example"
   17
       bundleVersion = "1.0.0"
   18
   )
   19
   20
   type bundle struct{}
   21
   22
   var _ v1.BundleProvider = (*bundle)(nil)
   23
   24
   func (b *bundle) GetBundleVersion() v1.BundleInfo {
   25
       return v1.BundleInfo{Name: bundleName, Version: &bundleVersion}
   26
   }
   27
   28
   func (b *bundle) RegisterDags(dagbag v1.Registry) error {
   29
       d := dagbag.AddDag("go_task_syntax_example")
   30
       d.AddTask(transform)
   31
       return nil
   32
   }
   33
   34
   func main() {
   35
       if err := bundlev1server.Serve(&bundle{}); err != nil {
   36
           log.Fatal(err)
   37
       }
   38
   }
   39
   40
   func transform(ctx sdk.TIRunContext, client sdk.Client, logger *slog.Logger) (any, error) {
   41
       ti := ctx.TaskInstance()
   42
   43
       raw, err := client.GetXCom(ctx, ti.DagID, ti.RunID, "extract", nil, "return_value", nil)
   44
       if err != nil {
   45
           return nil, fmt.Errorf("reading extract XCom: %w", err)
   46
       }
   47
   48
       var payload struct {
   49
           Numbers []float64 `json:"numbers"`
   50
       }
   51
       encoded, err := json.Marshal(raw)
   52
       if err != nil {
   53
           return nil, fmt.Errorf("re-encoding extract payload: %w", err)
   54
       }
   55
       if err := json.Unmarshal(encoded, &payload); err != nil {
   56
           return nil, fmt.Errorf("decoding extract payload: %w", err)
   57
       }
   58
   59
       var sum float64
   60
       for _, n := range payload.Numbers {
   61
           sum += n
   62
       }
   63
   64
       logger.Info("summed numbers in Go", "sum", sum, "count", len(payload.Numbers))
   65
       return map[string]any{
   66
           "sum":         sum,
   67
           "count":       len(payload.Numbers),
   68
           "computed_by": "Go " + runtime.Version(),
   69
       }, nil
   70
   }

   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 1.24 or later to compile the task bundle.

1
$ go mod tidy
2
$ 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.

Step 4: Create the Dag

In your dags folder, create a file called go_task_syntax_example.py with the following code:

1
import random
2
3
from airflow.sdk import dag, task, chain
4
5
6
@dag(tags=["go sdk"])
7
def go_task_syntax_example():
8
    @task
9
    def extract():
10
        return {"numbers": [random.randint(1, 100) for _ in range(random.randint(3, 6))]}
11
12
    @task.stub(queue="golang")
13
    def transform(): ...
14
15
    @task
16
    def load(result):
17
        print(f"Go returned {result}")
18
        return result
19
20
    extracted = extract()
21
    transformed = transform()
22
    chain(extracted, transformed)
23
    load(transformed)
24
25
26
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.

Java SDK example

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.

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:

   1
   $ mkdir -p include/java_sdk/src/java/com/example/bundle
   2
   $ 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:

   org.gradle.configuration-cache=true
   projectVersion=1.0.0-SNAPSHOT

   settings.gradle:

   pluginManagement {
       repositories {
           maven {
               url "https://repository.apache.org/content/repositories/snapshots/"
               mavenContent { snapshotsOnly() }
           }
           gradlePluginPortal()
           mavenCentral()
       }
   }
   rootProject.name = "airflow-java-sdk-etl-example"

   build.gradle:

   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.

   1
   package com.example.bundle;
   2
   3
   import java.util.LinkedHashMap;
   4
   import java.util.List;
   5
   import java.util.Map;
   6
   import org.apache.airflow.sdk.*;
   7
   import org.slf4j.Logger;
   8
   import org.slf4j.LoggerFactory;
   9
   10
   @Builder.Dag(id = "java_task_syntax_example")
   11
   public class JavaEtlExample {
   12
     private static final Logger logger = LoggerFactory.getLogger(JavaEtlExample.class);
   13
   14
     @Builder.Task(id = "transform")
   15
     public Map<String, Object> transform(
   16
         Client client, @Builder.XCom(task = "extract") Map<String, Object> payload) {
   17
       logger.info("[transform/java] received payload from python 'extract' task: {}", payload);
   18
   19
       List<?> numbers = (List<?>) payload.get("numbers");
   20
       long sum = 0;
   21
       for (Object n : numbers) {
   22
         sum += ((Number) n).longValue();
   23
       }
   24
   25
       Map<String, Object> result = new LinkedHashMap<>();
   26
       result.put("sum", sum);
   27
       result.put("count", numbers.size());
   28
       result.put("computed_by", "Java " + System.getProperty("java.version"));
   29
   30
       logger.info("[transform/java] summed {} numbers to {}", numbers.size(), sum);
   31
       return result;
   32
     }
   33
   }

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.

   1
   package com.example.bundle;
   2
   3
   import java.util.List;
   4
   import org.apache.airflow.sdk.*;
   5
   6
   public class EtlBundleBuilder implements BundleBuilder {
   7
     @Override
   8
     public Iterable<Dag> getDags() {
   9
       return List.of(JavaEtlExampleBuilder.build());
   10
     }
   11
   12
     public static void main(String[] args) {
   13
       var bundle = new EtlBundleBuilder().build();
   14
       Server.create(args).serve(bundle);
   15
     }
   16
   }

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

1
$ gradle bundle
2
$ 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:

1
import random
2
3
from airflow.sdk import dag, task, chain
4
5
6
@dag(tags=["Java SDK"])
7
def java_task_syntax_example():
8
    @task
9
    def extract():
10
        return {"numbers": [random.randint(1, 100) for _ in range(random.randint(3, 6))]}
11
12
    @task.stub(queue="java")
13
    def transform(): ...
14
15
    @task
16
    def load(result):
17
        print(f"Java returned {result}")
18
        return result
19
20
    extracted = extract()
21
    transformed = transform()
22
    chain(extracted, transformed)
23
    load(transformed)
24
25
26
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.

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 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 for more information.