如何从DAG文件控制Airflow基础容器的K8s节点选择（亲和性/容忍度）？

核心原因 #

Airflow的KubernetesPodOperator默认会继承values.yaml中配置的全局Pod模板（包括亲和性、容忍度），要实现任务级别的节点调度，需要显式在operator中指定覆盖规则——任务级参数优先级高于全局配置。

解决方案 #

1. 直接在Operator中指定亲和性/容忍度 #

在KubernetesPodOperator实例中显式设置affinity和tolerations参数，这些参数会直接覆盖values.yaml中的全局配置：

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from datetime import datetime

with DAG(
    dag_id="task_on_specific_node",
    schedule_interval=None,
    start_date=datetime(2024, 1, 1),
) as dag:
    run_on_custom_node = KubernetesPodOperator(
        task_id="custom_node_task",
        name="custom-node-pod",
        image="your-business-image:v1",
        # 自定义节点亲和性，匹配标签为node-group=dag专属组的节点
        affinity={
            "nodeAffinity": {
                "requiredDuringSchedulingIgnoredDuringExecution": {
                    "nodeSelectorTerms": [
                        {
                            "matchExpressions": [
                                {
                                    "key": "node-group",
                                    "operator": "In",
                                    "values": ["dag-specific-node-group"]
                                }
                            ]
                        }
                    ]
                }
            }
        },
        # 自定义容忍度，允许调度到带有special-node=true污点的节点
        tolerations=[
            {
                "key": "special-node",
                "operator": "Equal",
                "value": "true",
                "effect": "NoSchedule"
            }
        ],
        namespace="airflow",
        get_logs=True,
        is_delete_operator_pod=True,
    )

2. 使用pod_override完全自定义Pod Spec #

如果需要更全面的Pod配置覆盖（比如除了亲和性/容忍度，还要修改资源限制、安全上下文等），可以使用pod_override参数直接传入自定义的Pod Spec对象：

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from kubernetes.client import V1Pod, V1PodSpec
from datetime import datetime

# 定义自定义Pod Spec
custom_pod_spec = V1Pod(
    spec=V1PodSpec(
        affinity={
            "nodeAffinity": {
                "preferredDuringSchedulingIgnoredDuringExecution": [
                    {
                        "weight": 100,
                        "preference": {
                            "matchExpressions": [
                                {
                                    "key": "node-type",
                                    "operator": "In",
                                    "values": ["high-memory"]
                                }
                            ]
                        }
                    }
                ]
            }
        },
        tolerations=[
            {
                "key": "high-memory-node",
                "operator": "Exists",
                "effect": "NoSchedule"
            }
        ],
        # 可添加其他Pod配置，比如资源限制
        containers=[
            {
                "name": "main-container",
                "image": "your-business-image:v1",
                "resources": {
                    "requests": {"memory": "8Gi", "cpu": "4"},
                    "limits": {"memory": "16Gi", "cpu": "8"}
                }
            }
        ]
    )
)

with DAG(
    dag_id="high_memory_task",
    schedule_interval=None,
    start_date=datetime(2024, 1, 1),
) as dag:
    run_on_high_memory_node = KubernetesPodOperator(
        task_id="high_memory_task",
        name="high-memory-pod",
        # 传入自定义Pod Spec覆盖全局配置
        pod_override=custom_pod_spec,
        namespace="airflow",
        get_logs=True,
        is_delete_operator_pod=True,
    )

3. 验证覆盖效果 #

部署DAG后，可通过Kubectl查看Pod的调度信息，确认是否匹配预期节点：

kubectl describe pod <pod-name> -n airflow

在输出的Node-Selectors和Tolerations部分，应该能看到DAG中配置的规则，而非values.yaml的全局设置。

内容的提问来源于stack exchange，提问作者John Snow