Airflow Celery Worker持续重复执行已完成任务问题求助

嘿，我来帮你搞定这个头疼的问题！这种情况我在维护Airflow集群时碰到过好几次，核心原因基本都是任务状态元数据不同步或者Celery Broker残留消息，咱们一步步来排查修复：

一、先清理Celery Broker的残留消息 #

这是最常见的原因——已完成的任务消息没从Broker里被正确移除，导致新Worker一启动就反复抓取这些“幽灵任务”。

• 先停掉出问题的Worker：

  airflow celery stop worker
  

• 清理对应独立队列的残留消息：
  • 如果用Redis当Broker：

    redis-cli DEL celery_你的队列名
    

  • 如果用RabbitMQ：

    rabbitmqctl purge_queue 你的队列名
    

• 重启Worker，指定监听你的独立队列：

  airflow celery start worker -q 你的队列名
  

二、检查Airflow元数据库的任务状态一致性 #

有时候Web界面显示任务已完成，但元数据库里的任务实例状态可能有异常，或者Worker读取元数据时出现延迟：

• 登录Airflow的元数据库（比如PostgreSQL），查询这些重复任务的状态：

  SELECT task_id, state FROM task_instance WHERE dag_id = '你的DAG名' AND execution_date = '任务执行时间';
  

• 如果发现状态不是success，手动更新为success：

  UPDATE task_instance SET state = 'success' WHERE task_id = '重复的任务ID' AND execution_date = '任务执行时间';
  

• 重启Airflow Scheduler，强制同步状态：

  airflow scheduler restart
  

三、调整Celery的消息确认配置 #

如果上面两步没解决，可能是Worker的消息确认机制出了问题，导致Broker一直认为任务没完成：

• 打开Airflow的airflow.cfg文件，找到Celery相关配置，调整以下参数：

  task_acks_late = True
  worker_prefetch_multiplier = 1
  

  • task_acks_late=True：Worker会在任务执行完成后才向Broker发送确认，避免中途崩溃导致消息重复
  • worker_prefetch_multiplier=1：限制Worker预取的消息数量，防止一次性拿太多消息导致状态不同步
• 重启所有Worker和Scheduler，让配置生效。

补充：从日志进一步定位 #

你提到旧任务日志有INFO级别的记录，建议你仔细查看日志里的task_instance_key_str字段，确认是不是同一个任务实例被反复执行。如果是，那基本就是上面的消息残留或状态同步问题；如果是不同实例，那可能是DAG的调度配置有问题（比如重复触发）。

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