airflow

airflow

用途

Airflow 是一款开源工作流编排工具，可将数据管道定义为代码，并对其进行调度和监控。它使用 Python 创建有向无环图（DAG），以表示任务依赖关系和执行流程。

适用场景

对于包含重复性数据任务的场景，例如 ETL 流程、批处理作业或具有依赖关系的复杂工作流，请使用 Airflow。当你需要动态调度、重试和监控数据工程管道，尤其是在使用 Spark 或数据库等工具进行生产级操作时，Airflow 是理想选择。

主要功能

• 使用 Python 将工作流定义为 DAG，指定任务、依赖关系和调度计划。
• 内置调度器，可按指定间隔（例如 cron 风格）运行任务。
• 提供 Web UI，可实时监控任务日志和 DAG 状态，还可通过 /admin/ 等端点查看。
• 提供 BashOperator 等运算符来运行 Shell 命令，或使用 PythonOperator 执行自定义函数。
• 提供可扩展的 Hook，用于集成其他工具，例如使用 PostgresHook 连接数据库。
• 通过 airflow.cfg 文件配置，例如设置 [core] executor = LocalExecutor 以进行本地测试。

使用模式

要使用 Airflow，请通过 pip install apache-airflow 安装，然后使用 airflow db init 初始化数据库。在 Airflow 主目录的 dags 文件夹中定义 DAG。请始终使用虚拟环境以避免冲突。进行身份验证时，可设置 $AIRFLOW_UID 等环境变量来隔离用户。

• 模式 1：对于定时 ETL，创建每天运行的 DAG，并使用传感器等待数据输入。
• 模式 2：使用依赖关系串联任务，例如，仅在数据库查询成功后运行 Python 脚本。
• 示例 1：定义一个简单的 DAG 来执行每日备份：

  from airflow import DAG
  from airflow.operators.bash import BashOperator
  dag = DAG('daily_backup', schedule_interval='@daily')
  task = BashOperator(task_id='backup', bash_command='mysqldump db > backup.sql', dag=dag)
  

• 示例 2：调度一个使用 Spark 处理数据的管道：

  from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
  task = SparkSubmitOperator(task_id='spark_job', application='/path/to/script.py', dag=dag)
  

常用命令/API

设置好环境后，从命令行运行 Airflow。如有需要，可使用 $AIRFLOW__CORE__FERNET_KEY 加密连接信息。

• CLI 命令：

  • 初始化数据库：airflow db init --with-db-init
  • 启动 Web 服务器：airflow webserver --port 8080
  • 运行调度器：airflow scheduler --dag-id my_dag
  • 触发 DAG：airflow dags trigger my_dag --conf '{"key":"value"}'
  • 列出 DAG：airflow dags list

• API 端点（通过 REST API 使用，在 airflow.cfg 中通过 [api] auth_backend = airflow.api.auth.backend.default 启用）：

  • GET /api/v1/dags 用于列出所有 DAG，需要使用设置为 $AIRFLOW_API_TOKEN 的 API 令牌进行身份验证。
  • POST /api/v1/dags/{dag_id}/dagRuns 用于触发 DAG 运行，例如使用 JSON 负载 {"conf": {"param": "value"}}。
  • 使用 requests 调用 API 的示例代码：

    import requests
    response = requests.get('http://localhost:8080/api/v1/dags', headers={'Authorization': f'Bearer {os.environ["AIRFLOW_API_TOKEN"]}'})
    print(response.json())
    

集成说明

通过 Hook 和运算符将 Airflow 与其他工具集成。对于密钥，请使用 Airflow 的 Variables 或 Connections，并将其存储在元数据库中。设置 $AIRFLOW_CONN_POSTGRES_DEFAULT 等环境变量来配置数据库连接（例如，postgresql://user:pass@localhost/db）。

• 与 Spark 集成：使用 SparkSubmitOperator，并在运算符中设置执行器配置，例如 conf={"spark.executor.memory": "4g"}。
• 与 AWS 集成：使用 S3Hook 执行文件操作；将 $AWS_ACCESS_KEY_ID 和 $AWS_SECRET_ACCESS_KEY 设置为环境变量。
• 对于 Kubernetes，在 airflow.cfg 中配置 [kubernetes] namespace = default，并使用 KubernetesPodOperator。

错误处理

在任务定义中配置重试来处理错误，例如 retries=3, retry_delay=timedelta(minutes=5)。通过 Web UI 或 airflow tasks logs <dag_id> <task_id> 查看日志。在 DAG 中使用 on_failure_callback 触发告警。

• 常见错误：由于依赖项导致任务失败；请确保数据库连接等前置条件已设置，以修复此问题。
• 操作步骤：在任务中添加 email_on_failure=True，并在 airflow.cfg 中设置 [smtp] smtp_host = your.smtp.server。
• 示例：定义一个包含错误处理的任务：

  from airflow.utils.email import send_email
  task = PythonOperator(task_id='failing_task', python_callable=my_function, on_failure_callback=lambda context: send_email('[email protected]', 'Task Failed', 'Error details'))
  

图关系

• 相关：spark（用于数据管道中的任务执行）、hadoop（用于分布式处理集成）和数据库工具（用于元数据存储）。
• 依赖：调度器组件和 postgres 或 s3 等外部钩子。
• 集成：数据工程集群中的编排工具，例如与 ETL 框架组合使用的工作流。