Run the KubernetesPodOperator on Astro

The KubernetesPodOperator is one of the most customizable Apache Airflow operators. A task using the KubernetesPodOperator runs in a dedicated, isolated Kubernetes Pod that terminates after the task completes. To learn more about the benefits and usage of the KubernetesPodOperator, see the KubernetesPodOperator Learn guide.

On Astro, the infrastructure required to run the KubernetesPodOperator is built into every Deployment and is managed by Astronomer. Astro supports setting a default Pod configuration so that any task Pods without specific resource requests and limits cannot exceed your expected resource usage for the Deployment.

Some task-level configurations will differ on Astro compared to other Airflow environments. Use this document to learn how to configure individual task Pods for different use cases on Astro. To configure the default Pod resources for all KubernetesPodOperator Pods, see Configure Kubernetes Pod resources.

Known limitations

• Cross-account service accounts are not supported on Pods launched in an Astro cluster. To allow access to external data sources, you can provide credentials and secrets to tasks.

• PersistentVolumes (PVs) are not supported on Pods launched in an Astro cluster.

• You can’t use an image built for an ARM architecture in the KubernetesPodOperator. To build images using the x86 architecture on a Mac with an Apple chip, include the --platform flag in the FROM command of the Dockerfile that constructs your custom image. For example:

  $
  FROM --platform=linux/amd64 postgres:latest

  If you use an ARM image, your KPO task will fail with the error: base] exec /usr/bin/psql: exec format error.

Prerequisites

• An Astro project.
• An Astro Deployment.

Set up the KubernetesPodOperator on Astro

The following snippet is the minimum configuration you’ll need to create a KubernetesPodOperator task on Astro:

1
from airflow.configuration import conf
2
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
3
4
namespace = conf.get("kubernetes", "NAMESPACE")
5
6
KubernetesPodOperator(
7
    namespace=namespace,
8
    image="<your-docker-image>",
9
    cmds=["<commands-for-image>"],
10
    arguments=["<arguments-for-image>"],
11
    labels={"<pod-label>": "<label-name>"},
12
    name="<pod-name>",
13
    task_id="<task-name>",
14
    get_logs=True,
15
    in_cluster=True,
16
)

For each instantiation of the KubernetesPodOperator, you must specify the following values:

• namespace = conf.get("kubernetes", "NAMESPACE"): Every Deployment runs on its own Kubernetes namespace within a cluster. Information about this namespace can be programmatically imported as long as you set this variable.
• image: This is the Docker image that the operator will use to run its defined task, commands, and arguments. Astro assumes that this value is an image tag that’s publicly available on Docker Hub. To pull an image from a private registry, see Pull images from a Private Registry.
• in_cluster: If a Connection object is not passed to the KubernetesPodOperator’s kubernetes_conn_id parameter, specify in_cluster=True to run the task in the Deployment’s Astro cluster.

Configure task-level Pod resources

Astro automatically allocates resources to Pods created by the KubernetesPodOperator. Unless otherwise specified in your task-level configuration, the amount of resources your task Pod can use is defined by your default Pod resource configuration. To further optimize your resource usage, Astronomer recommends specifying compute resource requests and limits for each task.

To do so, define a kubernetes.client.models.V1ResourceRequirements object and provide that to the container_resources argument of the KubernetesPodOperator. For example:

1
from airflow.configuration import conf
2
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
3
from kubernetes.client import models as k8s
4
5
compute_resources = k8s.V1ResourceRequirements(
6
    limits={"cpu": "800m", "memory": "3Gi"},
7
    requests={"cpu": "800m", "memory": "3Gi"}
8
)
9
10
namespace = conf.get("kubernetes", "NAMESPACE")
11
12
KubernetesPodOperator(
13
    namespace=namespace,
14
    image="<your-docker-image>",
15
    cmds=["<commands-for-image>"],
16
    arguments=["<arguments-for-image>"],
17
    labels={"<pod-label>": "<label-name>"},
18
    name="<pod-name>",
19
    container_resources=compute_resources,
20
    task_id="<task-name>",
21
    get_logs=True,
22
    in_cluster=True,
23
)

Applying the previous code example ensures that when this dag runs, it launches a Kubernetes Pod with exactly 800m of CPU and 3Gi of memory as long as that infrastructure is available in your Deployment. After the task finishes, the Pod will terminate gracefully.

Mount a temporary directory

To run a task run the KubernetesPodOperator that utilizes your Deployment’s ephemeral storage, mount an emptyDir volume to the KubernetesPodOperator. For example:

1
from airflow.configuration import conf
2
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
3
from kubernetes.client import models as k8s
4
5
volume = k8s.V1Volume(
6
    name="cache-volume",
7
    emptyDir={},
8
)
9
10
volume_mounts = [
11
    k8s.V1VolumeMount(
12
        mount_path="/cache", name="cache-volume"
13
    )
14
]
15
16
example_volume_test = KubernetesPodOperator(
17
    namespace=namespace,
18
    image="<your-docker-image>",
19
    cmds=["<commands-for-image>"],
20
    arguments=["<arguments-for-image>"],
21
    labels={"<pod-label>": "<label-name>"},
22
    name="<pod-name>",
23
    task_id="<task-name>",
24
    get_logs=True,
25
    in_cluster=True,
26
    volume_mounts=volume_mounts,
27
    volumes=[volume],
28
)

Run the current Airflow image

You can run your Deployment’s current Airflow image with the KubernetesPodOperator by using the environment variable, ASTRONOMER_AIRFLOW_IMAGE. This environment variable allows you to run KPO tasks using the same Runtime image that your Deployment uses, in a way that won’t be affected by underlying cluster infrastructure changes that might change the base image repository URL. The ASTRONOMER_AIRFLOW_IMAGE environment variable allows you to ensure that your Deployment retrieves the correct image URL. The following code example shows how you can configure your KPO with ASTRONOMER_AIRFLOW_IMAGE:

1
import os
2
from airflow.configuration import conf
3
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
4
from kubernetes.client import models as k8s
5
KubernetesPodOperator(
6
    namespace=conf.get("kubernetes", "NAMESPACE"),
7
    image=os.getenv("ASTRONOMER_AIRFLOW_IMAGE"),
8
    image_pull_secrets=[
9
        k8s.V1LocalObjectReference("image-pull-secret"),
10
        k8s.V1LocalObjectReference("per-dp-registry-pull-secret"),
11
    ],
12
    cmds=["<commands-for-image>"],
13
    arguments=["<arguments-for-image>"],
14
    labels={"<pod-label>": "<label-name>"},
15
    name="<pod-name>",
16
    task_id="<task-name>",
17
    get_logs=True,
18
    in_cluster=True,
19
)

Run images from a private registry

By default, the KubernetesPodOperator expects to pull a Docker image that’s hosted publicly. If your images are hosted on the container registry native to your cloud provider, you can grant access to the images directly. Otherwise, if you are using any other private registry, you need to create a Kubernetes Secret containing credentials to the registry, then specify the Kubernetes Secret in your dag.

1
{
2
    "auths": {
3
        "https://index.docker.io/v1/": {
4
            "auth": "c3R...zE2"
5
        }
6
    }
7
}

1
from kubernetes.client import models as k8s
2
3
KubernetesPodOperator(
4
    namespace=namespace,
5
    image_pull_secrets=[k8s.V1LocalObjectReference("<your-secret-name>")],
6
    image="<your-docker-image>",
7
    cmds=["<commands-for-image>"],
8
    arguments=["<arguments-for-image>"],
9
    labels={"<pod-label>": "<label-name>"},
10
    name="<pod-name>",
11
    task_id="<task-name>",
12
    get_logs=True,
13
    in_cluster=True,
14
)

Use secret environment variables with the KubernetesPodOperator

Astro environment variables marked as secrets are stored in a Kubernetes secret called env-secrets. To use a secret value in a task running on the Kubernetes executor, you pull the value from env-secrets and mount it to the Pod running your task as a new Kubernetes Secret.

1. Add the following import to your dag file:

   1
   from airflow.kubernetes.secret import Secret

2. Define a Kubernetes Secret in your dag instantiation using the following format:

   1
   secret_env = Secret(deploy_type="env", deploy_target="<VARIABLE_KEY>", secret="env-secrets", key="<VARIABLE_KEY>")
   2
   namespace = conf.get("kubernetes", "NAMESPACE")

3. Reference the key for the environment variable, formatted as $VARIABLE_KEY in the task using the KubernetesPodOperator.

In the following example, a secret named MY_SECRET is pulled from env-secrets and printed to logs.

1
import pendulum
2
from airflow.kubernetes.secret import Secret
3
4
from airflow.models import DAG
5
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
6
from airflow.configuration import conf
7
8
with dag(
9
        dag_id='test-kube-pod-secret',
10
        start_date=pendulum.datetime(2022, 1, 1, tz="UTC"),
11
        end_date=pendulum.datetime(2022, 1, 5, tz="UTC"),
12
        schedule_interval="@once",
13
) as dag:
14
15
    secret_env = Secret(deploy_type="env", deploy_target="MY_SECRET", secret="env-secrets", key="MY_SECRET")
16
17
    namespace = conf.get("kubernetes", "NAMESPACE")
18
19
    k = KubernetesPodOperator(
20
        namespace=namespace,
21
        image="ubuntu:16.04",
22
        cmds=["bash", "-cx"],
23
        arguments=["echo $MY_SECRET && sleep 150"],
24
        name="test-name",
25
        task_id="test-task",
26
        get_logs=True,
27
        in_cluster=True,
28
        secrets=[secret_env],
29
    )

Use the @task.kubernetes decorator

The @task.kubernetes decorator provides a TaskFlow alternative to the traditional KubernetesPodOperator, which allows you to run a specified task in its own Kubernetes pod. Note that the Docker image provided to the @task.kubernetes decorator’s image parameter must support executing Python scripts in order to leverage the KubernetesPodOperator decorator.

Like regular @task decorated functions, XComs can be passed to the Python script running in the dedicated Kubernetes pod. If do_xcom_push is set to True in the decorator parameters, the value returned by the decorated function is pushed to XCom.

Astronomer recommends using the @task.kubernetes decorator instead of the KubernetesPodOperator when using XCom with Python scripts in a dedicated Kubernetes pod.

1
from pendulum import datetime
2
from airflow.configuration import conf
3
from airflow.decorators import dag, task
4
import random
5
6
# get the current Kubernetes namespace Airflow is running in
7
namespace = conf.get("kubernetes", "NAMESPACE")
8
9
@dag(
10
    start_date=datetime(2023, 1, 1),
11
    catchup=False,
12
    schedule="@daily",
13
)
14
def kubernetes_decorator_example_dag():
15
    @task
16
    def extract_data():
17
        # simulating querying from a database
18
        data_point = random.randint(0, 100)
19
        return data_point
20
21
    @task.kubernetes(
22
        # specify the Docker image to launch, it needs to be able to run a Python script
23
        image="python",
24
        # launch the Pod on the same cluster as Airflow is running on
25
        in_cluster=True,
26
        # launch the Pod in the same namespace as Airflow is running in
27
        namespace=namespace,
28
        # Pod configuration
29
        # naming the Pod
30
        name="my_pod",
31
        # log stdout of the container as task logs
32
        get_logs=True,
33
        # log events in case of Pod failure
34
        log_events_on_failure=True,
35
        # enable pushing to XCom
36
        do_xcom_push=True,
37
    )
38
    def transform(data_point):
39
        multiplied_data_point = 23 * int(data_point)
40
        return multiplied_data_point
41
42
    @task
43
    def load_data(**context):
44
        # pull the XCom value that has been pushed by the KubernetesPodOperator
45
        transformed_data_point = context["ti"].xcom_pull(
46
            task_ids="transform", key="return_value"
47
        )
48
        print(transformed_data_point)
49
50
    load_data(transform(extract_data()))
51
52
53
kubernetes_decorator_example_dag()

Related documentation

• How to use cluster ConfigMaps, Secrets, and Volumes with Pods
• KubernetesPodOperator Airflow Guide