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_datedays_ago(1)和schedule_interval0 2 * * *明确界定了执行窗口retries3, retry_delaytimedelta(minutes5)直接把容错策略写进代码里。这不是炫技是把过去靠文档、靠口头约定、靠人肉记忆的调度逻辑全部收归到 Git 仓库里由 CI/CD 流水线自动校验、部署、回滚。所以选 Airflow首要原因从来不是“它能做什么”而是“它让协作变得简单让故障变得可追溯”。2.2 Executor 选型Sequential、Local、Celery、Kubernetes到底该用哪个Airflow 的 Executor 是它的“心脏”决定了任务如何被分发和执行。选错 Executor后面所有优化都是白搭。我画了一张实际压测对比表这是我们在金融客户生产环境跑满三个月的真实数据Executor 类型单节点最大并发数任务启动延迟P95故障隔离性运维复杂度适用场景Sequential1 100ms无极低本地开发、单机 DemoLocal~32受 CPU/内存限制200-500ms弱进程级低小型团队、测试环境、百级任务/天Celery数千取决于 Worker 数量800ms - 2s强进程网络隔离中高需维护 Redis/RabbitMQ Worker 集群中大型企业、千级任务/天、需要异构执行环境如部分任务跑在 WindowsKubernetes理论无限Pod 弹性伸缩1.5s - 5s含 Pod 启动最强容器级隔离、资源配额高需 K8s 集群、Operator 经验云原生架构、任务负载波动大、严格资源管控、多租户我的建议非常明确新手起步无脑选 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 引入了序列化 DAGSerialized 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_dateDAG 第一次可被调度的时间点注意不是“现在开始”而是“从这个时间点起按 schedule 往后推”、retries默认重试次数schedule_interval是灵魂参数它决定 DAG 的触发节奏。daily等价于0 0 * * *但前者更语义化hourly等价于0 * * * *如果你想每 15 分钟跑一次必须写*/15 * * * *而不能写0,15,30,45 * * * *因为后者在某些时区下会有偏差。血肉推荐项决定健壮性catchupFalse这是生产环境的保命开关。默认为True意味着如果你今天才部署一个start_datedays_ago(30)的 DAGAirflow 会立刻补跑过去 30 天的所有历史实例瞬间压垮你的数据库和下游服务。设为False它只从部署时刻起按 schedule 往后调度。max_active_runs1限制同一个 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_varlocalhost) db_port int(Variable.get(prod_db_port, default_var5432)) # 2. 定义默认参数所有 Task 都继承这些值 default_args { owner: data_engineering, depends_on_past: False, # 不依赖前一个周期是否成功避免雪崩 start_date: datetime(2024, 1, 1), # DAG 生效的最早时间点 email: [alertcompany.com], email_on_failure: True, email_on_retry: False, retries: 2, retry_delay: timedelta(minutes5), execution_timeout: timedelta(hours2), # 全局超时防止单个任务卡死 } # 3. 创建 DAG 实例注意变量名必须是 dag dag DAG( dag_idetl__user_behavior_daily, default_argsdefault_args, description每日用户行为宽表生成, schedule_interval0 2 * * *, # 每天凌晨 2 点 catchupFalse, # 关键生产环境必须为 False max_active_runs1, # 关键防止单个 DAG 并发失控 tags[etl, user, daily], # 4. 使用 Jinja2 模板动态注入日期 user_defined_macros{ ds_add: lambda ds, days: (datetime.strptime(ds, %Y-%m-%d) timedelta(daysdays)).strftime(%Y-%m-%d) } ) # 5. 定义任务使用 PythonOperator 封装核心逻辑 def extract_user_data(**context): 从 Kafka 拉取昨日数据存入临时表 execution_date context[ds] # Airflow 自动注入的执行日期 # 这里写你的 PySpark 或 Pandas 代码 print(fExtracting data for {execution_date}) def transform_user_behavior(**context): 清洗、关联、聚合生成宽表 execution_date context[ds] # 这里写你的 SQL 或 DataFrame 操作 print(fTransforming data for {execution_date}) def load_to_warehouse(**context): 将宽表加载到数仓供 BI 查询 execution_date context[ds] # 这里写你的 INSERT INTO ... SELECT ... print(fLoading data for {execution_date}) # 6. 实例化 Operator并用 定义依赖 extract_task PythonOperator( task_idextract_user_data, python_callableextract_user_data, dagdag, ) transform_task PythonOperator( task_idtransform_user_behavior, python_callabletransform_user_behavior, dagdag, ) load_task PythonOperator( task_idload_to_warehouse, python_callableload_to_warehouse, dagdag, ) # 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没做幂等retries3结果一个失败的 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。状态层面善用XComsCross-Communication让上游 Task 把关键结果如“本次处理了 123456 条记录”xcom_push给下游下游可以用xcom_pull拿到用于决策或告警。BashOperator里加echo PROCESSED_ROWS123456 /tmp/xcom.json然后在下游PythonOperator里context[task_instance].xcom_pull(keyreturn_value)就能拿到。4. 实操过程与核心环节实现详解4.1 从零部署基于 Docker Compose 的生产级最小可行环境Airflow 官方推荐的生产部署方式是 Helm ChartK8s或 systemdLinux。但对于绝大多数中小团队Docker Compose 是最快、最可控、最易调试的起点。下面是我维护了两年、在 7 个客户现场成功上线的docker-compose.yml它不是一个玩具而是一个可直接用于小规模生产的精简版version: 3 services: # 1. PostgreSQL 元数据库 postgres: image: postgres:13 environment: - POSTGRES_USERairflow - POSTGRES_PASSWORDairflow - POSTGRES_DBairflow 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_EXn - FERNET_KEY46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho - AIRFLOW__CORE__EXECUTORCeleryExecutor - AIRFLOW__CORE__SQL_ALCHEMY_CONNpostgresqlpsycopg2://airflow:airflowpostgres/airflow - AIRFLOW__CELERY__RESULT_BACKENDdbpostgresql://airflow:airflowpostgres/airflow - AIRFLOW__CELERY__BROKER_URLredis://:redis:6379/1 - AIRFLOW__CORE__FERNET_KEY46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho - AIRFLOW__CORE__DAGS_FOLDER/opt/airflow/dags - AIRFLOW__CORE__LOGGING_LEVELINFO 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 adminexample.com airflow webserver # 4. Airflow Scheduler核心大脑负责 DAG 解析和任务派发 scheduler: build: . restart: always depends_on: - webserver environment: - LOAD_EXn - FERNET_KEY46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho - AIRFLOW__CORE__EXECUTORCeleryExecutor - AIRFLOW__CORE__SQL_ALCHEMY_CONNpostgresqlpsycopg2://airflow:airflowpostgres/airflow - AIRFLOW__CELERY__RESULT_BACKENDdbpostgresql://airflow:airflowpostgres/airflow - AIRFLOW__CELERY__BROKER_URLredis://:redis:6379/1 - AIRFLOW__CORE__FERNET_KEY46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho - AIRFLOW__CORE__DAGS_FOLDER/opt/airflow/dags - AIRFLOW__CORE__LOGGING_LEVELINFO 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_EXn - FERNET_KEY46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho - AIRFLOW__CORE__EXECUTORCeleryExecutor - AIRFLOW__CORE__SQL_ALCHEMY_CONNpostgresqlpsycopg2://airflow:airflowpostgres/airflow - AIRFLOW__CELERY__RESULT_BACKENDdbpostgresql://airflow:airflowpostgres/airflow - AIRFLOW__CELERY__BROKER_URLredis://:redis:6379/1 - AIRFLOW__CORE__FERNET_KEY46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho - AIRFLOW__CORE__DAGS_FOLDER/opt/airflow/dags - AIRFLOW__CORE__LOGGING_LEVELINFO 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 KeyFERNET_KEY是 Airflow 加密敏感信息如 Connection 密码的密钥。它必须是 32 字节的 base64 编码字符串。上面的46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho是官方示例生产环境必须用openssl rand -base64 32生成一个全新的、保密的 Key并写入所有服务的环境变量中。Key 一旦设定就不能再改否则所有加密的密码都会失效。airflow db upgrade这是最关键的初始化命令。它会根据当前 Airflow 版本自动执行数据库迁移migration创建所有必需的表结构。webserver启动时必须先执行它否则会报错。airflow users create一键创建管理员账号省去 Web UI 里手动注册的麻烦。部署流程只有三步mkdir airflow-project cd airflow-projectmkdir dags logs pluginsdocker-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.pyimport unittest from airflow.models import DagBag class TestMyDag(unittest.TestCase): def setUp(self): self.dagbag DagBag(dag_folder/path/to/your/dags, include_examplesFalse) def test_dag_loads_with_no_errors(self): 验证 DAG 文件语法正确能被 Airflow 成功加载 self.assertFalse( len(self.dagbag.import_errors), fDAG 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 新手遇到的最高频问题。我把它整理成一张速查表按发生概率从高到低排序排查步骤检查项如何验证解决方案1. 文件位置与命名DAG 文件是否在dags_folder目录下文件名是否以.py结尾docker-compose exec webserver ls /opt/airflow/dags/确保文件在正确路径且后缀是.py不能是.py.txt2. 变量名DAG 实例的变量名是否为dag打开 DAG 文件看dag DAG(...)这一行必须是dag DAG(...), 不能是my_dag DAG(...)3. 语法错误DAG 文件是否有 Python 语法错误docker-compose exec webserver python /opt/airflow/dags/your_dag.py修复所有SyntaxError比如少括号、冒号、缩进错误4. Import 错误是否导入了不存在的模块docker-compose logs webserver | grep ImportErrorpip install缺失的包或检查requirements.txt5.start_date问题start_date是否设置得过于“未来”docker-compose exec webserver airflow dags list-import-errors将start_date改为一个过去的日期如datetime(2023, 1, 1)最有效的排查命令是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(chunksize10000)代替一次性读取。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