Airflow问题求助：变量未赋值错误及CeleryExecutor任务排队异常

我之前也踩过Airflow的这些坑，分享下我的解决思路和实际验证过的方案：

这个错误本质是Python变量作用域问题，90%以上的情况是你自定义的Operator（比如PythonOperator、自定义BaseOperator子类）里的代码存在路径分支未初始化变量的情况。举个典型的错误场景：

def process_data(**context):
    if context["dag_run"].conf.get("use_custom_file"):
        filename = context["dag_run"].conf["file_path"]
    # 这里如果上面的条件不成立，filename就没被定义，执行下面的代码就会报错
    with open(filename, "r") as f:
        data = f.read()

解决步骤： #

• 补全变量初始化逻辑：确保在任何代码路径下，filename都被提前赋值。比如给默认值或者补全else分支：

  def process_data(**context):
      # 先给默认值兜底
      filename = "/path/to/default_file.txt"
      if context["dag_run"].conf.get("use_custom_file"):
          filename = context["dag_run"].conf["file_path"]
      with open(filename, "r") as f:
          data = f.read()
  

• 排查Airflow版本bug：如果是使用Airflow内置Operator出现这个错误，可能是特定版本的bug，建议升级到Airflow 2.x的稳定小版本（比如2.8.x或更高），这类小问题通常会被官方修复。

这个问题我遇到过好几次，核心原因通常是调度器、Worker和Redis之间的通信/资源瓶颈，以下是按优先级排序的排查和解决方向：

1. 先排查Redis的状态 #

Redis作为Celery的broker和结果后端，是最容易出问题的环节：

• 检查Redis日志，看有没有OOM command not allowed（内存不足）的报错：如果有，修改Redis配置文件的maxmemory（比如调到4GB）和maxmemory-policy（设置为allkeys-lru），然后重启Redis。
• 用redis-cli info stats查看队列指标：关注connected_clients（连接数）、keyspace_hits/keyspace_misses（缓存命中率），如果连接数接近Redis的maxclients限制，需要调整Redis的maxclients参数，或者优化Airflow的Celery连接池配置（在airflow.cfg里设置celery__broker_pool_limit，比如设为100）。

2. 检查Worker的资源和日志 #

• 资源瓶颈排查：用top/htop查看Worker进程的CPU、内存占用，如果Worker频繁被OOM杀死（系统日志里会有Out of memory记录），要么增加Worker的资源配额，要么降低airflow.cfg里的celery__worker_concurrency（建议设置为Worker所在机器CPU核心数的1-2倍，比如4核机器设为6）。
• 开启DEBUG级日志：启动Worker时用airflow celery worker -l DEBUG，比--raw能拿到更详细的任务处理日志。我之前遇到过任务里用了lambda函数，Celery用pickle序列化失败，导致任务一直卡在队列里，换成普通def定义的函数就解决了。

3. 优化调度器配置 #

如果调度器没法及时把任务推送到队列，也会导致任务堆积：

• 调整airflow.cfg里的调度器参数：
  • scheduler__num_parallel_executors：增加并行执行的调度器进程数（比如设为8）
  • scheduler__max_threads：增加调度器的线程数（比如设为16）
• 检查调度器日志，排除数据库（比如PostgreSQL/MySQL）的性能问题（比如数据库连接池不足、查询慢），这也会间接导致调度延迟。

4. 验证Celery队列状态 #

用Airflow的Celery命令查看队列和Worker状态：

• airflow celery inspect active：查看Worker正在处理的任务
• airflow celery inspect reserved：查看Worker已经领取但还没开始处理的任务
• airflow celery inspect stats：查看Worker的统计信息，比如任务处理速率

如果发现某个Worker的reserved任务一直不减少，说明这个Worker可能卡住了，单独重启该Worker即可，不用全量重启。

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