实测 | 工作流搭建之Composer 2

在实际部署 Composer 2 的时候，我见过很多同学因为配置错误导致整个工作流瘫痪。最常见的是在定义 DAG 时，忘记设置 `schedule_interval` 或者 `start_date`，这样会导致任务永远不会被触发。另外，很多人用 `bash` 命令来执行 Python 脚本，但没注意 `bash` 默认不支持 `.py` 文件的执行权限，结果运行失败。还有人把 `default_args` 写在了循环里，导致每次任务实例化时都重新定义，触发大量日志浪费。关键点在于，必须确保 `trigger_rule` 和 `upstream` 的逻辑正确，否则任务会卡在依赖未满足的状态里。Composer 2 的 `max_active_runs` 设置影响并发数量，但很多人没意识到它与 `concurrency` 是两个不同的参数，从而导致资源竞争或任务堆积。最值得说的还是它的 `retry` 和 `max_retries` 配置，这两个参数在任务失败时能自动重试，但如果你设置成 `retry=3`，那么需要确保 `max_retries` 也够，否则会触发任务终止。

▌ 技术参考

Composer 2 是 Airflow 的一个分支，专门用于团队协作和分布式部署。它引入了 `CeleryExecutor` 和 `KubernetesExecutor`，让任务调度更加灵活。在搭建流程中，需要先确保安装 Python 3.7+，然后通过 pip 安装 Composer 2，比如 `pip install apache-airflow-composer==2.0.0`。安装完成后，要配置 `airflow.cfg` 文件，主要涉及 `executor`, `sql_alchemy_conn`, `dags_folder` 和 `plugins_folder`，这些参数决定了 airflow 如何连接数据库、存储 DAG 以及加载插件。需要注意的是，`sql_alchemy_conn` 必须是有效的数据库连接字符串，否则会触发启动失败。

搭建 Composer 2 时，数据库的选择非常关键。大多数用户会使用 PostgreSQL，因为它在并发性能和数据一致性上表现优于 SQLite。如果使用 PostgreSQL，需要确保 `airflow` 用户有权限创建数据库，同时设置正确的 `host`, `port`, `user`, `password` 和 `database`。另外，DAG 文件路径需要在 `dags_folder` 下，比如 `/home/airflow/dags`，并且每个 DAG 必须是一个 `.py` 文件，不能是 `.yaml` 或 `.json` 格式。如果 DAG 文件写错了，比如没有正确导入 `DAG` 类，任务将无法被识别。

在定义 DAG 时，必须使用 `DAG` 类并设置 `default_args`，其中 `start_date` 和 `schedule_interval` 是必须的。比如 `default_args = {'start_date': datetime(2023, 1, 1), 'schedule_interval': '@daily'}`。如果 `start_date` 设置成未来时间，任务会延迟执行，但如果你设置错了，比如格式错误，会导致任务无法创建。另外，`schedule_interval` 有多种格式，`@daily` 每天执行一次，`0 0 ` 每天凌晨执行，`@weekly` 每周执行一次。在实际使用中，我见过不少用户混淆了这些格式，导致任务执行间隔不对。

运行 Composer 2 时，需要启动 `airflow scheduler` 和 `airflow worker` 两个进程。启动命令是 `airflow scheduler` 和 `airflow worker`，但很多人直接运行 `airflow webserver`，这样会导致 DAG 无法被调度。另外，启动 worker 时需要指定 `--pool` 参数，比如 `airflow worker --pool default`，否则任务会分配到错误的 executor。如果 worker 启动后没有响应，需要检查 `airflow.cfg` 中的 `executor` 配置是否正确，比如是否使用 `CeleryExecutor` 或 `KubernetesExecutor`，以及是否有足够的 worker 节点。

在 DAG 中定义 tasks 时，必须使用 `PythonOperator` 或 `BashOperator`，并且每个 task 需要有一个唯一的 `task_id`。如果两个 task 的 `task_id` 重复，会导致编译错误。此外，使用 `PythonOperator` 时，`python_callable` 必须是函数名，不能是类的实例。我见过一些用户直接传入函数对象，结果导致执行失败。另外，`op_kwargs` 可以传递参数，但需要确保格式正确，比如 `{'param1': 'value1', 'param2': 'value2'}`。

任务依赖关系是 Composer 2 的核心，必须使用 `set_upstream` 或 `set_downstream` 来定义。比如 `task1 >> task2` 表示 task1 成功后执行 task2。如果依赖关系写错了，比如用 `<<` 而不是 `>>`，任务会永远停留在未执行状态。此外，`trigger_rule` 的设置也非常关键，比如 `all_success` 表示所有上游任务成功才会继续执行，而 `one_success` 表示只要有一个成功就能继续。在实际使用中，我见过一些用户没有设置 `trigger_rule`，导致任务无法正确触发。

Composer 2 的 `max_active_runs` 参数影响并发任务数量，必须根据实际负载合理设置。如果设置成 5，那么每个 DAG 最多同时运行 5 个实例。此外，`concurrency` 参数影响每个 worker 的并发能力，比如 `concurrency=10` 表示一个 worker 最多同时处理 10 个任务。这两个参数不能混淆，否则会导致任务堆积或资源不足。在生产环境中，建议将 `max_active_runs` 设置为 `None`，让 airflow 自动管理。

任务日志是调试的关键，必须配置 `log_filename` 和 `log_file_prefix`。比如 `log_filename = 'task.log'` 和 `log_file_prefix = 'task_'`，这样日志会更清晰。此外，日志保留时间可以通过 `log_rotation_days` 设置，比如 `log_rotation_days = 7` 表示保留 7 天的日志。我见过一些用户没设置这些参数，导致日志堆积太多，占用大量磁盘空间，这也会影响调度性能。

在使用 `KubernetesExecutor` 时，需要配置 `kubernetes_conn_id` 和 `kubernetes_namespace`。比如 `kubernetes_conn_id = 'k8s_default'` 和 `kubernetes_namespace = 'default'`。如果这些配置错误，会导致任务无法在 Kubernetes 上执行。此外，`kubernetes_executor_config` 可以设置额外参数，比如 `image` 和 `resources`，确保任务容器有足够资源。在实际部署中，我见过很多用户直接使用默认配置，结果任务在 Kubernetes 上崩溃或资源不足。

任务重试机制是 Composer 2 的一个强大功能，但配置不当会引发很多问题。`retry` 参数设置的是重试次数，`max_retries` 是最大重试次数。如果 `retry=3` 而 `max_retries=2`，任务最多只能重试两次。此外，`retry_delay` 控制每次重试的时间间隔，比如 `retry_delay=datetime.timedelta(minutes=5)`，这样任务会在失败后等待 5 分钟再重试。在一些高并发场景下，重试策略需要根据任务类型调整，比如网络请求任务可能需要更长的间隔。

Composer 2 的 DAG 调度依赖 `scheduler` 和 `worker` 两个进程，这两个进程必须在不同机器上运行，否则无法实现分布式调度。如果两者运行在同台服务器，会导致资源竞争，影响性能。此外，任务执行过程中需要确保 `executor` 有权限访问存储任务文件的目录，比如 `/home/airflow/dags`。如果权限不够，会触发错误，导致任务无法执行。

在使用 `BashOperator` 时，必须确保命令路径正确，否则会执行失败。比如 `bash_command = 'python /home/airflow/dags/script.py'`，如果路径错误，任务会直接挂掉。此外，`BashOperator` 会自动处理环境变量，但需要确保这些变量在执行环境中存在，否则会出问题。在一些 CI/CD 场景下，我见过用户忘记设置环境变量，导致 script 执行失败。

Composer 2 的 DAG 文件需要符合 Python 语法，否则无法被解析。比如 `from datetime import datetime` 必须放在文件开头，否则会触发错误。另外，DAG 文件不能有语法错误，比如缺少冒号或括号，否则无法被加载。在实际使用中，我见过一些用户直接复制粘贴 DAG 内容，但没有检查语法，导致任务无法启动。

任务依赖链需要合理设计，避免环形依赖。比如 `task1 >> task2 >> task1` 会导致调度失败，因为任务无法完成。此外，`set_upstream` 和 `set_downstream` 需要正确使用，确保任务之间的依赖关系清晰。在一些复杂的业务逻辑中，我见过开发者把依赖链写得过于复杂，导致任务无法正确触发。

Composer 2 的 `max_active_runs` 和 `max_concurrent_runs` 是两个不同的参数，前者限制每个 DAG 最多同时运行的任务数，后者限制每个 worker 的并发能力。如果只设置 `max_active_runs=5`，但 `max_concurrent_runs=10`，那么整个 DAG 可能会超过限制。在实际部署中，我见过不少用户直接忽略了这两个参数的区别，导致任务堆积或资源不足。

在使用 `CeleryExecutor` 时，需要确保 `celery_broker_url` 和 `celery_result_backend` 配置正确。比如 `celery_broker_url = 'redis://localhost:6379/0'` 和 `celery_result_backend = 'redis://localhost:6379/0'`。如果这两个参数设置错误，会导致任务无法被发送或接收。此外，需要确保 Redis 服务正常运行，否则整个任务调度会失败。在一些中小型团队中，我见过他们直接使用本地 Redis，结果任务执行时出错。

任务失败后，Composer 2 默认会记录失败原因，但需要配置 `on_failure_callback` 来定义失败后的处理逻辑。比如 `on_failure_callback = my_failure_handler`，这样可以及时通知运维人员。此外，`on_success_callback` 和 `on_retry_callback` 也可以自定义，用于任务成功或重试后的处理。在实际项目中，我见过一些用户没有配置这些回调函数，导致任务失败后无人知晓。