如何基于KubernetesPodOperator任务Pod指标实现Airflow CeleryExecutor动态扩缩容

你遇到的这个问题确实很典型——CeleryExecutor下的Airflow Worker本身只是调度KubernetesPodOperator（KPO）的任务Pod，自身几乎不占用CPU/内存资源，所以常规的基于Worker资源使用率的HPA完全起不到作用。不过别担心，还是有几种可行的办法可以实现基于任务负载的Worker动态扩缩容：

方案一：基于自定义指标的HorizontalPodAutoscaler（推荐） #

核心思路是用任务队列积压量或正在运行的KPO任务Pod数量作为扩缩容触发指标，而不是Worker自身的资源。要实现这个，需要借助Prometheus采集Airflow和K8s的指标，再通过Prometheus Adapter把这些指标暴露给K8s HPA。

具体步骤： #

1. 开启Airflow Metrics
   如果你用的是官方Airflow Helm Chart，直接在values.yaml里开启Metrics：

   metrics:
     enabled: true
     serviceMonitor:
       enabled: true # 方便Prometheus自动发现指标
   

   这样Airflow会暴露Celery队列长度指标airflow_celery_queue_length，这是判断任务积压的核心指标。

2. 部署Prometheus和Prometheus Adapter

   • 部署Prometheus用于采集Airflow和K8s的Pod指标
   • 配置Prometheus Adapter，将Airflow的队列长度指标转化为K8s可识别的自定义指标。比如在Adapter的配置中添加：

     rules:
     - seriesQuery: 'airflow_celery_queue_length{queue!=""}'
       resources:
         overrides:
           namespace: {resource: "namespace"}
       name:
         matches: 'airflow_celery_queue_length'
         as: 'airflow_queue_length'
       metricsQuery: 'sum(airflow_celery_queue_length{queue="$queue"}) by (namespace)'
     

   配置完成后，K8s API就能获取到airflow_queue_length这个自定义指标。

3. 配置HPA基于自定义指标扩缩容
   创建HPA资源，以Worker Deployment为目标，基于队列长度或KPO任务Pod数量调整副本数：

   apiVersion: autoscaling/v2
   kind: HorizontalPodAutoscaler
   metadata:
     name: airflow-worker-hpa
   spec:
     scaleTargetRef:
       apiVersion: apps/v1
       kind: Deployment
       name: airflow-worker
     minReplicas: 2 # 最小Worker数
     maxReplicas: 10 # 最大Worker数
     metrics:
     # 基于Celery队列长度扩容
     - type: External
       external:
         metric:
           name: airflow_queue_length
           selector:
             matchLabels:
               queue: "default" # 你的目标队列名称
         target:
           type: Value
           value: 10 # 队列长度超过10时触发扩容
     # 可选：基于正在运行的KPO任务Pod数量扩容
     - type: Pods
       pods:
         metric:
           name: kube_pod_status_ready
         selector:
           matchLabels:
             airflow-task-type: "kubernetes-pod-operator" # 替换为你的KPO任务Pod标签
         target:
           type: AverageValue
           averageValue: 5 # 平均每个Worker对应5个运行中的任务Pod
     # 配置扩缩容行为避免频繁抖动
     behavior:
       scaleUp:
         stabilizationWindowSeconds: 300
         policies:
         - type: Percent
           value: 50
           periodSeconds: 60
       scaleDown:
         stabilizationWindowSeconds: 600
         policies:
         - type: Percent
           value: 30
           periodSeconds: 60
   

方案二：自定义监控脚本/Operator #

如果你不想引入Prometheus生态，也可以写一个简单的脚本，定期获取Airflow队列长度或K8s中KPO任务Pod的数量，然后直接调用K8s API调整Worker Deployment的副本数。

示例思路： #

• 用Python的kubernetes库连接K8s集群
• 用Airflow的REST API或Celery的celery inspect active命令获取队列长度
• 设置阈值逻辑：比如队列长度>15时扩容2个Worker，队列长度<3时缩容1个Worker
• 把这个脚本做成CronJob（比如每分钟执行一次）或者常驻Deployment

方案三：切换到KubernetesExecutor（可选） #

你提到HPA更适配KubernetesExecutor，确实如此——KubernetesExecutor下每个任务直接由K8s调度，Worker（实际是Scheduler）不需要常驻，扩缩容逻辑更直接。但如果因为业务原因必须保留CeleryExecutor，这个方案可以作为长期优化方向。

注意事项： #

• 扩缩容时要考虑Worker的启动时间，设置合理的stabilizationWindowSeconds避免频繁扩缩
• 确保Worker的资源配置足够支撑任务分发（虽然KPO任务不跑在Worker上，但Worker需要处理任务调度逻辑）
• 如果使用多个Celery队列，要针对每个队列单独配置指标和HPA规则

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