Apache Airflow 是一个基于 Python 的工作流编排平台，用于构建、调度和监控数据管道、ETL/ELT 流程及定时任务。本技能（Skill）旨在帮助用户掌握 Airflow 的工作流编排能力，覆盖 DAG 编写、operator、sensor、hook、XCom、变量、连接以及部署模式等内容。
适用场景：当需要构建具有任务依赖关系的复杂数据管道、编排 ETL/ELT 工作流、调度周期性批处理任务、管理带重试与错误处理的工作流、协调跨多个系统的任务、获得工作流执行历史的可见性、需要审计追踪与数据血缘、或构建机器学习管道编排时，可使用本技能。
不适用场景：实时流式数据处理（应使用 Kafka、Flink）、简单 cron 作业（应使用 systemd 定时器、crontab）、CI/CD 管道（应使用 GitHub Actions、Jenkins）、低延迟需求（Airflow 存在调度器开销）、单任务简单自动化（过于笨重）、面向非开发者的可视化工作流设计（应使用 n8n）。
安装前置条件与方式，资料提供三种：
Option 1：pip（开发环境）。创建虚拟环境并设置 AIRFLOW_HOME，使用约束文件安装指定版本（示例为 2.8.1）。安装命令包含： python -m venv airflow-env source airflow-env/bin/activate export AIRFLOW_HOME=~/airflow AIRFLOW_VERSION=2.8.1 PYTHON_VERSION="$(python --version | cut -d " " -f 2 | cut - "." -f 1-2)" CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt" pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}" 随后初始化数据库：airflow db init；创建管理员用户 airflow users create（含 username、firstname、lastname、role、email、password 参数）；启动服务 airflow webserver --port 8080 与 airflow scheduler。
Option 2：Docker Compose（推荐）。下载官方 docker-compose.yaml（地址为 https://airflow.apache.org/docs/apache-airflow/2.8.1/docker-compose.yaml），创建 ./dags ./logs ./plugins ./config 目录，写入 .env 中的 AIRFLOW_UID，执行 docker compose up airflow-init 初始化，再 docker compose up -d 启动，UI 访问地址为 http://localhost:8080（airflow/airflow）。
Option 3：Kubernetes with Helm。添加 apache-airflow 仓库、更新仓库，使用 helm install 安装并指定 --namespace airflow、--create-namespace、--set executor=KubernetesExecutor，通过 kubectl get secret 获取 web UI 密码。
开发环境配置：pip install apache-airflow[dev,postgres,celery,kubernetes]；测试工具 pip install pytest pytest-airflow；代码检查 pip install ruff。
核心能力包括四部分。
-
基础 DAG 结构：通过 default_args 配置 owner、depends_on_past、email、email_on_failure、email_on_retry、retries、retry_delay、retry_exponential_backoff、max_retry_delay、execution_timeout 等；DAG 定义包含 dag_id、default_args、description、schedule_interval（如 '0 6 * * *'）、start_date、catchup、max_active_runs、tags、doc_md。使用 PythonOperator、BashOperator、EmptyOperator，任务间通过 >> 定义依赖；Python 任务通过 context 获取执行上下文（如 ds），返回值经 XCom 传递，Bash 模板中可使用 {{ ds }}、{{ ti.xcom_pull(...) }}。
-
高级 operator：BranchPythonOperator 实现条件分支；TaskGroup 进行逻辑分组（示例含 validate_schema、validate_quality、validate_completeness 并行任务）；动态任务映射（Airflow 2.3+）使用 PythonOperator.partial(...).expand(...)；使用 trigger_rule（如 TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS）控制汇合行为。
-
Sensor 事件驱动模式：FileSensor（filepath、poke_interval、timeout、mode、soft_fail）、S3KeySensor（bucket_name、bucket_key、aws_conn_id、mode='reschedule'）、HttpSensor（http_conn_id、endpoint、response_check）、ExternalTaskSensor（external_dag_id、external_task_id、execution_delta、allowed_states、failed_states）、PythonSensor（自定义条件）。多个 sensor 全部通过后再进入处理任务。
-
Hook 外部系统集成：PostgresHook（get_conn、get_pandas_df、执行 SQL 并返回结果）、S3Hook（load_string 上传数据）、HttpHook、SlackWebhookHook。示例展示从 PostgreSQL 抽取、再上传 S3 的流程。
关于 Airflow 版本支持范围、各部署方式的资源要求、各 provider 包的完整列表、定价与收费信息、性能基准、社区支持渠道等，资料暂未提供。