Configure task-level Pod resources | Astronomer Documentation

Setup

Define container resources

Define a kubernetes.client.models.V1ResourceRequirements object and provide that to the container_resources argument of the KubernetesPodOperator. For example:

The following 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 terminates gracefully.

1 from airflow.configuration import conf
2 from airflow.providers.cncf.kubernetes.operators.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 )

When you add custom labels, do not remove or modify the default Airflow labels that Astro applies to KPO Pods. See Known limitations.

On Astro Hosted, Astro automatically sets resource requests equal to limits for KubernetesPodOperator task 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.

1	from airflow.configuration import conf
2	from airflow.providers.cncf.kubernetes.operators.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	)