Configure tasks to run with the Kubernetes executor

The Kubernetes executor runs each Airflow task in a dedicated Kubernetes Pod. On Astro, you can customize these Pods on a per-task basis using a pod_override configuration. If a task doesn’t contain a pod_override configuration, it runs using the default Pod as configured in your Deployment resource settings.

This document describes how to configure individual task Pods for different use cases. To configure defaults for all Kubernetes executor task pods, see Configure Kubernetes Pod resources.

Prerequisites

An Astro Deployment using Astro Runtime version 8.1.0 or later.

If you use the Kubernetes executor on Astro, you can’t change the PYTHONPATH of your Astro project from its default value. If you do, the Kubernetes executor will be unable to read airflow_local_settings.py and will fail to start up new Pods.

Customize a task’s Kubernetes Pod

By default, Astro supports a maximum KubernetesExecutor Pod size of 43 vCPU and 86 GiB of memory. Astro can support Pod sizes up to 86 vCPU and 172 GiB of memory on request. Contact Astro support if you need to run larger jobs on KubernetesExecutor.

While you can customize all values for a worker Pod, Astronomer does not recommend configuring complex Kubernetes infrastructure in your Pods, such as sidecars. These configurations have not been tested by Astronomer.

For each task running with the Kubernetes executor, you can customize its individual worker Pod and override the defaults used in Astro by configuring a pod_override file.

Add the following import to your dag file:

from kubernetes.client import models as k8s

Add a pod_override configuration to the dag file containing the task. See the kubernetes-client GitHub for a list of all possible settings you can include in the configuration.
Specify the pod_override in the task’s parameters.

See the following example of a pod_override configuration.

Example: Set CPU or memory limits and requests

You can request a specific amount of resources for a Kubernetes worker Pod so that a task always has enough resources to run successfully. When requesting resources, make sure that your requests don’t exceed the resource limits in your Deployment’s max pod size. The following example shows how you can use a pod_override configuration in your dag code to request custom resources for a task:

import pendulum
import time

from airflow.models.dag import DAG
from airflow.decorators import task
from airflow.operators.bash import BashOperator
from airflow.example_dags.libs.helper import print_stuff
from kubernetes.client import models as k8s

k8s_exec_config_resource_requirements = {
    "pod_override": k8s.V1Pod(
        spec=k8s.V1PodSpec(
            containers=[
                k8s.V1Container(
                    name="base",
                    resources=k8s.V1ResourceRequirements(
                        requests={"cpu": 0.5, "memory": "1024Mi", "ephemeral-storage": "1Gi"},
                        limits={"cpu": 0.5, "memory": "1024Mi", "ephemeral-storage": "1Gi"}
                    )
                )
            ]
        )
    )
}

with DAG(
    dag_id="example_kubernetes_executor_pod_override_sources",
    schedule=None,
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    catchup=False
):
    BashOperator(
      task_id="bash_resource_requirements_override_example",
      bash_command="echo hi",
      executor_config=k8s_exec_config_resource_requirements
    )

    @task(executor_config=k8s_exec_config_resource_requirements)
    def resource_requirements_override_example():
        print_stuff()
        time.sleep(60)

    resource_requirements_override_example()

When this dag runs, it launches a Kubernetes Pod with exactly 0.5m of CPU and 1024Mi of memory, as long as that infrastructure is available in your Deployment. After the task finishes, the Pod terminates gracefully.

On Astro Hosted, Astro automatically sets resource requests equal to limits for KubernetesExecutor worker Pods. This ensures Pods receive a Kubernetes Guaranteed Quality of Service (QoS) class, which prevents resource contention and eviction. Because Astro uses the limit values as both requests and limits, your Pods are billed based on the limits you configure, even if actual usage is lower. To avoid unexpected charges, set limits close to the resources your task requires. Check your Billing and usage to view your resource use and associated charges.

Use secret environment variables in worker Pods

On Astro Deployments, secret environment variable values are stored in a Kubernetes secret called env-secrets. These environment variables are available to your worker Pods, and you can access them in your tasks just like any other environment variable. For example, you can use os.environ[<your-secret-env-var-key>] or os.getenv(<your-secret-env-var-key>, None) in your dag code to access the variable value. However, if you can’t use Python, or you are using a pre-defined code that expects specific keys for environment variables, you must pull the secret value from env-secrets and mount it to the Pod running your task as a new Kubernetes Secret.

Add the following import to your dag file:

from airflow.kubernetes.secret import Secret

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

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

Specify the Secret in the secret_key_ref section of your pod_override configuration.
In the task where you want to use the secret value, add the following task-level argument:
op_kwargs={ "env_name": secret_env.deploy_target },
In the executable for the task, call the secret value using os.environ[env_name].

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

import pendulum
from kubernetes.client import models as k8s

from airflow.configuration import conf
from airflow.kubernetes.secret import Secret
from airflow.models import DAG
from airflow.operators.python import PythonOperator


def print_env(env_name):
    import os
    print(os.environ[env_name])

with DAG(
        dag_id="test-secret",
        start_date=pendulum.datetime(2022, 1, 1, tz="UTC"),
        end_date=pendulum.datetime(2022, 1, 5, tz="UTC"),
        schedule="@once",
) as dag:
    secret_env = Secret(deploy_type="env", deploy_target="MY_SECRET", secret="env-secrets", key="MY_SECRET")
    namespace = conf.get("kubernetes", "NAMESPACE")

    p = PythonOperator(
        python_callable=print_env,
        op_kwargs={
            "env_name": secret_env.deploy_target
        },
        task_id="test-py-env",
        executor_config={
            "pod_override": k8s.V1Pod(
                spec=k8s.V1PodSpec(
                    containers=[
                        k8s.V1Container(
                            name="base",
                            env=[
                                k8s.V1EnvVar(
                                    name=secret_env.deploy_target,
                                    value_from=k8s.V1EnvVarSource(
                                        secret_key_ref=k8s.V1SecretKeySelector(name=secret_env.secret,
                                                                               key=secret_env.key)
                                    ),
                                )
                            ],
                        )
                    ]
                )
            ),
        }
    )

Overview

Get started

Develop DAGs

AI features

Deploy code

Manage Deployments

Automation & CI/CD

Platform observability

Administration

Release notes

Best practices

Reference

Airflow 3

Configure tasks to run with the Kubernetes executor

Prerequisites

Customize a task’s Kubernetes Pod

Example: Set CPU or memory limits and requests

Use secret environment variables in worker Pods

See also

​Prerequisites

​Customize a task’s Kubernetes Pod

​Example: Set CPU or memory limits and requests

​Use secret environment variables in worker Pods

​See also

Prerequisites

Customize a task’s Kubernetes Pod

Example: Set CPU or memory limits and requests

Use secret environment variables in worker Pods

See also