Airflow生产级调度系统从零搭建实战指南

1. 项目概述：从零开始搭建一个真正能用的 Airflow 生产级调度系统

你是不是也经历过这样的场景：写好了一段 Python 数据清洗脚本，又配好了数据库连接，本地跑通了，结果一到定时执行就抓瞎？ crontab 配半天，日志找不到，失败了没通知，依赖关系全靠脑补，改个时间还得 SSH 进服务器手动编辑——这哪是做数据工程，这是在给服务器当人肉闹钟。我第一次在电商公司搭调度系统时，就是这么过来的，整整两周卡在“怎么让三个脚本按顺序跑、中间出错自动停、成功了发钉钉、失败了立刻电话告警”这个基本需求上。直到我把 Apache Airflow 真正吃透，才明白它根本不是什么高大上的“AI 工具”，而是一套把“任务调度”这件事彻底工程化、可视化、可追溯的基础设施。它解决的不是某个算法问题，而是每天都在发生的、最基础也最要命的“自动化流水线”问题。关键词里提到的 Towards AI，其实恰恰说明了当前行业的一个真实趋势：Airflow 已经从数据团队的专属工具，变成了 AI 工程师、MLOps 工程师、甚至后端开发都必须掌握的通用能力。它不绑定 Python，但用 Python 写起来最顺；它不强制要求 Kubernetes，但在 K8s 上跑得最稳；它看起来配置复杂，但核心逻辑就三句话：任务（Task）是什么、谁来跑（Executor）、什么时候跑（Schedule）。这篇文章，就是我过去三年在五家不同规模公司落地 Airflow 的实战笔记。不讲虚的原理图，不堆概念，只告诉你：从pip install开始，到第一个 DAG 在 Web UI 上绿色运行，再到它真正在你公司的生产环境里扛住每天百万级任务调度，每一步踩过什么坑、为什么这么选、参数背后的真实含义是什么。如果你正打算搭一个调度系统，或者手里的 Airflow 总是莫名其妙挂掉、日志查不到、DAG 不刷新，那这篇就是为你写的。

2. 整体架构设计与方案选型逻辑拆解

2.1 为什么不是 Cron、Supercronic 或其他轻量级方案？

很多人第一反应是：“我一个脚本，crontab 一行命令搞定，为啥要搞 Airflow 这么重？”这个问题我被问过不下五十次。答案不在功能多寡，而在协作成本和故障成本。举个真实例子：我们曾有一个每日凌晨 2 点跑的用户行为宽表生成任务，最初用 crontab，后来业务方提需求，要加一个“如果上游数据延迟超过 30 分钟，就跳过本次执行”。于是运维同学在 crontab 里加了一段 shell 脚本去查 HDFS 时间戳，再加判断。又过了两周，另一个团队说他们那个报表依赖这个宽表，希望宽表跑完立刻触发他们的计算。于是又加了一个curl调用他们 API 的逻辑。再后来，DBA 发现这个脚本占用了太多内存，要求降级，于是又加了ulimit和资源限制……最后那个 crontab 条目长得像一首诗，没人敢动，每次上线新版本都要全组 review。而 Airflow 的 DAG 文件，本质上就是一份可版本控制、可 Code Review、可单元测试的调度契约。task_a >> task_b >> task_c这一行代码，清晰表达了三个任务的强依赖；start_date=days_ago(1)和schedule_interval='0 2 * * *'明确界定了执行窗口；retries=3, retry_delay=timedelta(minutes=5)直接把容错策略写进代码里。这不是炫技，是把过去靠文档、靠口头约定、靠人肉记忆的调度逻辑，全部收归到 Git 仓库里，由 CI/CD 流水线自动校验、部署、回滚。所以选 Airflow，首要原因从来不是“它能做什么”，而是“它让协作变得简单，让故障变得可追溯”。

2.2 Executor 选型：Sequential、Local、Celery、Kubernetes，到底该用哪个？

Airflow 的 Executor 是它的“心脏”，决定了任务如何被分发和执行。选错 Executor，后面所有优化都是白搭。我画了一张实际压测对比表，这是我们在金融客户生产环境跑满三个月的真实数据：

我的建议非常明确：新手起步，无脑选 Local；公司已有 K8s，直接上 KubernetesExecutor；其他所有情况，优先考虑 CeleryExecutor。为什么？因为 Local 模式下，Scheduler 和 Worker 全在一个进程里，没有网络开销，调试极其方便，airflow tasks test my_dag my_task 2024-01-01命令能秒级反馈，是理解 DAG 执行生命周期的最佳沙盒。而 Celery 是目前社区最成熟、文档最全、问题排查路径最清晰的分布式方案。我见过太多团队一开始贪图 Kubernetes 的“先进”，结果卡在 Pod 初始化超时、Sidecar 注入失败、ServiceAccount 权限配置上，耽误一个月上线。反观 Celery，哪怕你只有一台 8C16G 的云服务器，装上 Redis 和几个 Worker，就能轻松支撑每天 5000+ 任务。它的核心优势在于“解耦”：Scheduler 只管编排和派发，Worker 只管执行，两者通过消息队列通信。这意味着你可以把 CPU 密集型任务（比如模型训练）的 Worker 单独部署在 GPU 机器上，把 IO 密集型任务（比如数据库导出）的 Worker 部署在离数据库近的机器上，这种灵活性是 Local 和 Kubernetes（初期）都难以比拟的。

2.3 数据库选型：PostgreSQL 为何是唯一推荐？

Airflow 必须依赖一个外部数据库来存储元数据（DAG 定义、任务状态、历史记录、用户信息等）。官方支持 SQLite、MySQL、PostgreSQL。但 SQLite 是开发玩具，生产环境绝对禁止；MySQL 曾经是主流，但自 Airflow 2.0 起，官方明确将 PostgreSQL 列为首选和事实标准。原因有三：第一，事务一致性。Airflow 的 Scheduler 是一个多线程组件，它需要频繁地对task_instance、dag_run、serialized_dag等表进行高并发的SELECT FOR UPDATE和INSERT ... ON CONFLICT操作。PostgreSQL 的 MVCC（多版本并发控制）机制在处理这类混合读写负载时，比 MySQL 的行锁表现稳定得多，极少出现死锁。第二，JSONB 字段支持。Airflow 2.0 引入了序列化 DAG（Serialized DAG）机制，把整个 DAG 对象序列化成 JSON 存进数据库。PostgreSQL 的JSONB类型原生支持索引、查询、部分更新，而 MySQL 的JSON类型在 5.7 版本中性能孱弱，索引支持不完善。第三，扩展生态。像pg_stat_statements这样的插件，能帮你精准定位慢查询，比如发现SELECT * FROM task_instance WHERE state = 'running'这条语句耗时飙升，立刻就知道是某个 DAG 的任务堆积了，而不是在一堆慢日志里大海捞针。我建议的最小安全配置是：PostgreSQL 12+，开启shared_preload_libraries = 'pg_stat_statements'，并为task_instance表的dag_id,state,execution_date字段建立复合索引。这套组合拳下来，即使你的任务量涨到每天十万级，元数据库依然稳如泰山。

3. 核心细节解析与实操要点精讲

3.1 DAG 文件结构：不只是 Python，而是一份“契约”

一个.py文件被 Airflow 识别为 DAG，需要满足几个硬性条件，缺一不可。很多新手卡在这里，以为文件名对了就行，结果 Web UI 里死活不显示 DAG。我把它拆解成“骨架”和“血肉”两部分：

骨架（必须项）：

• 必须定义一个DAG类的实例，且变量名不能是dag以外的任何名字（比如my_dag就不行，Airflow 解析器只认dag这个变量名）；
• DAG构造函数中，dag_id参数必须是全局唯一的字符串，不能包含空格或特殊字符，最佳实践是project_name__pipeline_name__version（如etl__user_behavior_v2），双下划线是 Airflow 官方推荐的命名分隔符；
• default_args字典必须包含owner（负责人）、start_date（DAG 第一次可被调度的时间点，注意不是“现在开始”，而是“从这个时间点起，按 schedule 往后推”）、retries（默认重试次数）；
• schedule_interval是灵魂参数，它决定 DAG 的触发节奏。'@daily'等价于'0 0 * * *'，但前者更语义化；'@hourly'等价于'0 * * * *'；如果你想每 15 分钟跑一次，必须写'*/15 * * * *'，而不能写'0,15,30,45 * * * *'，因为后者在某些时区下会有偏差。

血肉（推荐项，决定健壮性）：

• catchup=False：这是生产环境的保命开关。默认为True，意味着如果你今天才部署一个start_date=days_ago(30)的 DAG，Airflow 会立刻补跑过去 30 天的所有历史实例，瞬间压垮你的数据库和下游服务。设为False，它只从部署时刻起，按 schedule 往后调度。
• max_active_runs=1：限制同一个 DAG 同时最多只能有一个运行中的DagRun。防止上游数据延迟导致多个周期的任务堆积，把下游打垮。
• tags=['etl', 'critical']：给 DAG 打标签，Web UI 里可以按标签筛选，运维时价值巨大。

下面是一个我在线上环境使用的、经过千锤百炼的 DAG 模板，每一行都有其存在的理由：

from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.bash import BashOperator from airflow.models import Variable # 1. 从 Airflow Variable 中读取敏感配置，而非硬编码 db_host = Variable.get("prod_db_host", default_var="localhost") db_port = int(Variable.get("prod_db_port", default_var="5432")) # 2. 定义默认参数，所有 Task 都继承这些值 default_args = { 'owner': 'data_engineering', 'depends_on_past': False, # 不依赖前一个周期是否成功，避免雪崩 'start_date': datetime(2024, 1, 1), # DAG 生效的最早时间点 'email': ['alert@company.com'], 'email_on_failure': True, 'email_on_retry': False, 'retries': 2, 'retry_delay': timedelta(minutes=5), 'execution_timeout': timedelta(hours=2), # 全局超时，防止单个任务卡死 } # 3. 创建 DAG 实例，注意变量名必须是 'dag' dag = DAG( dag_id='etl__user_behavior_daily', default_args=default_args, description='每日用户行为宽表生成', schedule_interval='0 2 * * *', # 每天凌晨 2 点 catchup=False, # 关键！生产环境必须为 False max_active_runs=1, # 关键！防止单个 DAG 并发失控 tags=['etl', 'user', 'daily'], # 4. 使用 Jinja2 模板，动态注入日期 user_defined_macros={ 'ds_add': lambda ds, days: (datetime.strptime(ds, '%Y-%m-%d') + timedelta(days=days)).strftime('%Y-%m-%d') } ) # 5. 定义任务，使用 PythonOperator 封装核心逻辑 def extract_user_data(**context): """从 Kafka 拉取昨日数据，存入临时表""" execution_date = context['ds'] # Airflow 自动注入的执行日期 # 这里写你的 PySpark 或 Pandas 代码 print(f"Extracting data for {execution_date}") def transform_user_behavior(**context): """清洗、关联、聚合，生成宽表""" execution_date = context['ds'] # 这里写你的 SQL 或 DataFrame 操作 print(f"Transforming data for {execution_date}") def load_to_warehouse(**context): """将宽表加载到数仓，供 BI 查询""" execution_date = context['ds'] # 这里写你的 INSERT INTO ... SELECT ... print(f"Loading data for {execution_date}") # 6. 实例化 Operator，并用 >> 定义依赖 extract_task = PythonOperator( task_id='extract_user_data', python_callable=extract_user_data, dag=dag, ) transform_task = PythonOperator( task_id='transform_user_behavior', python_callable=transform_user_behavior, dag=dag, ) load_task = PythonOperator( task_id='load_to_warehouse', python_callable=load_to_warehouse, dag=dag, ) # 7. 用位运算符定义执行顺序：extract -> transform -> load extract_task >> transform_task >> load_task

提示：context是 Airflow 传递给每个任务的上下文字典，里面包含了ds（执行日期，格式YYYY-MM-DD）、ts（时间戳，格式YYYY-MM-DDTHH:MM:SS）、dag_run（当前 DagRun 对象）等关键信息。务必学会用context['ds']替代datetime.now().strftime('%Y-%m-%d')，否则你的任务永远跑的是“当前时间”，而不是“调度时间”。

3.2 Task 设计哲学：原子性、幂等性、可观测性

Airflow 的 Task 不是“一段代码”，而是一个有生命周期、有状态、可重试、可监控的独立单元。设计一个好 Task，有三个铁律：

第一，原子性（Atomicity）：一个 Task 必须代表一个不可再分的最小业务动作。比如，“从 S3 下载文件”是一个原子 Task，“解压文件”是另一个，“解析 CSV”是第三个。绝不能把下载、解压、解析全写在一个PythonOperator里。为什么？因为一旦解压失败，Airflow 会重试整个 Task，导致重复下载，浪费带宽；而如果分开了，下载成功后，解压失败，就只重试解压，效率高，也便于定位问题。

第二，幂等性（Idempotency）：同一个 Task，在相同输入参数下，无论执行多少次，结果都必须一致。这是重试机制的生命线。实现幂等最常用的方法是“先查后写”或“覆盖写”。例如，你的load_to_warehouseTask，不要写INSERT INTO table VALUES (...)，而应该写INSERT INTO table SELECT ... FROM temp_table ON CONFLICT (id) DO UPDATE SET ...（PostgreSQL）或REPLACE INTO table ...（MySQL）。对于文件操作，写入前先检查目标路径是否存在，存在则删除，再写入。我见过最惨的案例，是一个同事写了一个“发送邮件”的 Task，没做幂等，retries=3，结果一个失败的 DAG 触发了 3 封一模一样的告警邮件，老板的邮箱直接被塞爆。

第三，可观测性（Observability）：一个 Task 必须能被轻易地“看到”它在做什么、做到哪一步、为什么失败。这体现在三个层面：日志、指标、状态。日志层面，每个 Task 开头打印Starting task at {datetime.now()}，结尾打印Task completed successfully，中间关键步骤（如“已读取 100 万行”、“已写入 50 万条”）都要打点。指标层面，用airflow.providers.prometheus插件暴露airflow_task_duration_seconds_count等指标，接入你的 Prometheus/Grafana。状态层面，善用XComs（Cross-Communication），让上游 Task 把关键结果（如“本次处理了 123456 条记录”）xcom_push给下游，下游可以用xcom_pull拿到，用于决策或告警。BashOperator里加echo "PROCESSED_ROWS=123456" >> /tmp/xcom.json，然后在下游PythonOperator里context['task_instance'].xcom_pull(key='return_value')就能拿到。

4. 实操过程与核心环节实现详解

4.1 从零部署：基于 Docker Compose 的生产级最小可行环境

Airflow 官方推荐的生产部署方式是 Helm Chart（K8s）或 systemd（Linux）。但对于绝大多数中小团队，Docker Compose 是最快、最可控、最易调试的起点。下面是我维护了两年、在 7 个客户现场成功上线的docker-compose.yml，它不是一个玩具，而是一个可直接用于小规模生产的精简版：

version: '3' services: # 1. PostgreSQL 元数据库 postgres: image: postgres:13 environment: - POSTGRES_USER=airflow - POSTGRES_PASSWORD=airflow - POSTGRES_DB=airflow volumes: - postgres-db-volume:/var/lib/postgresql/data healthcheck: test: ["CMD-SHELL", "pg_isready -U airflow -d airflow"] interval: 30s timeout: 10s retries: 5 # 2. Redis 消息队列（CeleryExecutor 必需） redis: image: redis:7-alpine expose: - 6379 healthcheck: test: ["CMD", "redis-cli", "ping"] interval: 30s timeout: 10s retries: 5 # 3. Airflow Webserver：提供 Web UI 和 API webserver: build: . restart: always depends_on: - postgres - redis environment: - LOAD_EX=n - FERNET_KEY=46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho= - AIRFLOW__CORE__EXECUTOR=CeleryExecutor - AIRFLOW__CORE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow - AIRFLOW__CELERY__RESULT_BACKEND=db+postgresql://airflow:airflow@postgres/airflow - AIRFLOW__CELERY__BROKER_URL=redis://:@redis:6379/1 - AIRFLOW__CORE__FERNET_KEY=46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho= - AIRFLOW__CORE__DAGS_FOLDER=/opt/airflow/dags - AIRFLOW__CORE__LOGGING_LEVEL=INFO volumes: - ./dags:/opt/airflow/dags - ./logs:/opt/airflow/logs - ./plugins:/opt/airflow/plugins ports: - "8080:8080" command: >- bash -c " airflow db upgrade && airflow users create --username admin --password admin --firstname Admin --lastname User --role Admin --email admin@example.com && airflow webserver" # 4. Airflow Scheduler：核心大脑，负责 DAG 解析和任务派发 scheduler: build: . restart: always depends_on: - webserver environment: - LOAD_EX=n - FERNET_KEY=46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho= - AIRFLOW__CORE__EXECUTOR=CeleryExecutor - AIRFLOW__CORE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow - AIRFLOW__CELERY__RESULT_BACKEND=db+postgresql://airflow:airflow@postgres/airflow - AIRFLOW__CELERY__BROKER_URL=redis://:@redis:6379/1 - AIRFLOW__CORE__FERNET_KEY=46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho= - AIRFLOW__CORE__DAGS_FOLDER=/opt/airflow/dags - AIRFLOW__CORE__LOGGING_LEVEL=INFO volumes: - ./dags:/opt/airflow/dags - ./logs:/opt/airflow/logs - ./plugins:/opt/airflow/plugins command: airflow scheduler # 5. Airflow Worker：真正干活的苦力，可以水平扩展 worker: build: . restart: always depends_on: - redis - postgres environment: - LOAD_EX=n - FERNET_KEY=46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho= - AIRFLOW__CORE__EXECUTOR=CeleryExecutor - AIRFLOW__CORE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow - AIRFLOW__CELERY__RESULT_BACKEND=db+postgresql://airflow:airflow@postgres/airflow - AIRFLOW__CELERY__BROKER_URL=redis://:@redis:6379/1 - AIRFLOW__CORE__FERNET_KEY=46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho= - AIRFLOW__CORE__DAGS_FOLDER=/opt/airflow/dags - AIRFLOW__CORE__LOGGING_LEVEL=INFO volumes: - ./dags:/opt/airflow/dags - ./logs:/opt/airflow/logs - ./plugins:/opt/airflow/plugins command: airflow celery worker volumes: postgres-db-volume:

这个文件的关键设计点在于：

• 健康检查（healthcheck）：为postgres和redis都配置了healthcheck，确保webserver启动前，依赖服务已经 ready。没有这个，webserver会因连接不上数据库而反复崩溃。
• Fernet Key：FERNET_KEY是 Airflow 加密敏感信息（如 Connection 密码）的密钥。它必须是 32 字节的 base64 编码字符串。上面的46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho=是官方示例，生产环境必须用openssl rand -base64 32生成一个全新的、保密的 Key，并写入所有服务的环境变量中。Key 一旦设定，就不能再改，否则所有加密的密码都会失效。
• airflow db upgrade：这是最关键的初始化命令。它会根据当前 Airflow 版本，自动执行数据库迁移（migration），创建所有必需的表结构。webserver启动时必须先执行它，否则会报错。
• airflow users create：一键创建管理员账号，省去 Web UI 里手动注册的麻烦。

部署流程只有三步：

1. mkdir airflow-project && cd airflow-project
2. mkdir dags logs plugins
3. docker-compose up -d --build

等待 2 分钟，打开http://localhost:8080，用admin/admin登录，你就能看到一个干净的、可运行的 Airflow 环境了。这就是“最小可行产品”（MVP）的力量。

4.2 DAG 开发与调试：从本地测试到 Web UI 部署的完整闭环

Airflow 的开发体验，很大程度上取决于你能否快速验证代码。我总结了一套“三步验证法”，保证你的 DAG 从写完到上线，全程可控：

第一步：本地单元测试（Unit Test）在 DAG 文件同目录下，创建test_my_dag.py：

import unittest from airflow.models import DagBag class TestMyDag(unittest.TestCase): def setUp(self): self.dagbag = DagBag(dag_folder='/path/to/your/dags', include_examples=False) def test_dag_loads_with_no_errors(self): """验证 DAG 文件语法正确，能被 Airflow 成功加载""" self.assertFalse( len(self.dagbag.import_errors), f'DAG import errors: {self.dagbag.import_errors}' ) def test_dag_structure(self): """验证 DAG 的任务依赖关系是否符合预期""" dag = self.dagbag.get_dag('etl__user_behavior_daily') self.assertIsNotNone(dag) self.assertEqual(len(dag.tasks), 3) # 应该有 3 个 Task # 验证 extract_task 是 transform_task 的上游 self.assertIn('extract_user_data', [t.task_id for t in dag.get_task('transform_user_behavior').upstream_list]) if __name__ == '__main__': unittest.main()

运行python test_my_dag.py，如果输出OK，说明你的 DAG 文件没有语法错误，结构也符合预期。这是 CI/CD 流水线的第一道关卡。

第二步：本地任务测试（Task Test）这是最强大的调试手段。airflow tasks test命令会模拟一个完整的 Task 执行环境，但不触发调度器，不写入数据库，不调用外部服务，纯内存执行。假设你的 DAG 文件叫user_behavior_dag.py，放在./dags/目录下，那么：

# 进入 Airflow 容器 docker-compose exec webserver bash # 运行测试（注意：execution_date 必须是过去的时间，且符合 DAG 的 start_date） airflow tasks test etl__user_behavior_daily extract_user_data 2024-01-01

你会看到完整的日志输出，包括Starting attempt 1 of 1、Executing <TaskInstance: etl__user_behavior_daily.extract_user_data 2024-01-01 None [queued]>，以及你print()的所有内容。如果这里报错，100% 是你的 Python 代码逻辑问题，和 Airflow 环境无关。

第三步：Web UI 部署与手动触发（Trigger DAG）当你确认本地测试全部通过，就可以把 DAG 文件放进./dags/目录。Airflow 的 Scheduler 默认每 30 秒扫描一次dags_folder，发现新文件会自动加载。你可以在 Web UI 的DAGs页面，看到你的 DAG 出现，初始状态是Paused。点击右侧的Play按钮，将其置为Unpaused。然后点击 DAG 名称进入详情页，点击右上角的Trigger DAG按钮，选择一个Execution Date（比如2024-01-01），点击Trigger。这时，Scheduler 会创建一个DagRun，并按依赖关系依次将TaskInstance派发给 Worker 执行。你可以在Graph View里看到实时的执行流程图，在Tree View里看到所有任务的状态（Queued->Running->Success），在Log里看到每一行输出。这才是真正的“端到端”验证。

注意：Trigger DAG是手动触发，它会忽略schedule_interval，只运行一次。而Unpause后，DAG 才会按schedule_interval自动触发。很多新手混淆这两者，以为点了Trigger就万事大吉，结果第二天发现任务没跑，就是因为忘了Unpause。

5. 常见问题与排查技巧实录

5.1 DAG 不显示在 Web UI？九成是这五个原因

这是 Airflow 新手遇到的最高频问题。我把它整理成一张速查表，按发生概率从高到低排序：

最有效的排查命令是airflow dags list-import-errors，它会直接列出所有因语法或 Import 问题而加载失败的 DAG，并给出详细的错误堆栈。我建议把这个命令加入你的日常巡检清单。

5.2 任务卡在Queued状态？深度诊断指南

一个 Task 从None（未创建）到Queued（已排队）再到Running（正在执行），中间隔着 Scheduler、Broker、Worker 三个环节。卡在Queued，说明 Scheduler 已经把它派发出去了，但 Worker 没有拿到。原因通常有三类：

Broker 连接问题：这是最常见原因。Worker 无法连接到 Redis 或 RabbitMQ。验证方法：

# 进入 Worker 容器 docker-compose exec worker bash # 尝试连接 Redis redis-cli -h redis -p 6379 ping # 应该返回 PONG # 查看 Celery 的状态 celery -A airflow.executors.celery_executor.app status

如果status命令报错或返回0 nodes online，说明 Worker 没连上 Broker。检查AIRFLOW__CELERY__BROKER_URL环境变量是否拼写正确，redis服务名是否匹配。

Worker 进程未启动或崩溃：docker-compose ps查看worker服务的状态。如果是Restarting，说明 Worker 进程启动后立刻崩溃。查看日志：docker-compose logs worker。最常见的崩溃原因是ModuleNotFoundError，即 Worker 容器里缺少 DAG 代码依赖的包。解决方案是在Dockerfile中pip install -r requirements.txt，或者把requirements.txt挂载进去。

资源耗尽：Worker 的 CPU 或内存被打满，无法启动新进程。docker stats查看worker容器的资源使用率。如果MEM USAGE接近LIMIT，说明内存不足。解决方案是增加容器内存限制，或优化你的 Task 代码，减少内存占用（比如用pandas.read_csv(chunksize=10000)代替一次性读取）。

5.3 日志丢失或无法查看？终极解决方案

Airflow 的日志，默认存储在本地文件系统（/opt/airflow/logs）。这在单机开发时没问题，但一旦上了多 Worker 的生产环境，日志就分散在各个 Worker 机器上，查起来极其痛苦。我的解决方案是：强制所有日志统一输出到 stdout，并由 Docker 或 K8s 的日志收集器（如 Fluentd、Filebeat）统一采集到 ELK 或 Loki。

在docker-compose.yml的worker服务里，添加：

worker: # ... 其他配置 logging: driver: "json-file" options: max-size: "10m" max-file: "3" # ... 其他配置

同时，在airflow.cfg（或通过环境变量）中，关闭文件日志，强制使用控制台日志：

[logging] logging_level = INFO fab_logging_level = WARNING colored_log_format = True log_format = "[%(asctime)s] {%(filename)s:%(lineno)d} %(levelname)s - %(message)s" simple_log_format = "%(asctime)s %(levelname)s - %(message)s" # 关键：禁用文件日志 task_log_reader = console

这样，所有 Worker 的日志都会实时输出到docker-compose logs worker，