跳到主要内容

Airflow 与Prometheus集成

介绍​

Apache Airflow 是一个强大的工作流管理工具，广泛用于数据管道的编排和调度。然而，随着任务复杂性的增加，监控和日志管理变得至关重要。Prometheus 是一个开源的监控和告警工具，能够高效地收集和存储时间序列数据。通过将 Airflow 与 Prometheus 集成，您可以实时监控任务的执行情况，并快速定位问题。

本文将逐步介绍如何将 Airflow 与 Prometheus 集成，并提供实际案例和代码示例。

为什么需要集成？​

Airflow 本身提供了基本的监控功能，但在大规模任务调度场景下，这些功能可能不足以满足需求。Prometheus 提供了更强大的监控能力，包括：

  • 实时任务状态监控
  • 任务执行时间统计
  • 资源使用情况监控
  • 自定义告警规则

通过集成，您可以利用 Prometheus 的这些功能来增强 Airflow 的监控能力。

集成步骤​

1. 安装 Prometheus​

首先，您需要在您的环境中安装 Prometheus。您可以通过以下命令在本地安装 Prometheus：

bash
wget https://github.com/prometheus/prometheus/releases/download/v2.30.3/prometheus-2.30.3.linux-amd64.tar.gz
tar xvfz prometheus-2.30.3.linux-amd64.tar.gz
cd prometheus-2.30.3.linux-amd64
./prometheus --config.file=prometheus.yml

2. 配置 Airflow 以暴露指标​

Airflow 提供了一个内置的指标暴露机制，您可以通过配置 metrics 部分来启用它。在 airflow.cfg 文件中添加以下配置：

ini
[metrics]
statsd_on = True
statsd_host = localhost
statsd_port = 8125
statsd_prefix = airflow

3. 使用 Prometheus 收集 Airflow 指标​

Prometheus 可以通过 statsd_exporter 来收集 Airflow 的指标。首先，安装 statsd_exporter：

bash
wget https://github.com/prometheus/statsd_exporter/releases/download/v0.22.3/statsd_exporter-0.22.3.linux-amd64.tar.gz
tar xvfz statsd_exporter-0.22.3.linux-amd64.tar.gz
cd statsd_exporter-0.22.3.linux-amd64
./statsd_exporter --statsd.mapping-config=statsd_mapping.yml

然后，配置 statsd_mapping.yml 文件以映射 Airflow 的指标：

yaml
mappings:
- match: "airflow.*"
  name: "airflow_metrics"
  labels:
    job: "airflow"

4. 配置 Prometheus 以抓取 statsd_exporter​

在 prometheus.yml 文件中添加以下配置，以抓取 statsd_exporter 的数据：

yaml
scrape_configs:
  - job_name: 'statsd_exporter'
    static_configs:
      - targets: ['localhost:9102']

5. 启动服务并验证​

启动所有服务后，您可以通过 Prometheus 的 Web 界面（通常位于 http://localhost:9090）查看 Airflow 的指标。

实际案例​

假设您有一个 Airflow DAG，用于每天处理一批数据。您希望监控该 DAG 的执行时间和成功率。通过集成 Prometheus，您可以轻松实现以下目标：

  1. 监控任务执行时间：通过 Prometheus 的 histogram 指标类型，您可以记录每个任务的执行时间，并生成相应的图表。
  2. 设置告警规则：如果某个任务的执行时间超过预期，Prometheus 可以触发告警，通知您及时处理。

以下是一个简单的 DAG 示例：

python
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime

def process_data():
    # 模拟数据处理
    import time
    time.sleep(10)

dag = DAG('data_processing', description='Simple data processing DAG',
          schedule_interval='@daily',
          start_date=datetime(2023, 1, 1), catchup=False)

task = PythonOperator(task_id='process_data_task',
                      python_callable=process_data,
                      dag=dag)

通过 Prometheus，您可以监控 process_data_task 的执行时间，并设置相应的告警规则。

总结​

通过将 Airflow 与 Prometheus 集成，您可以显著增强任务监控和日志管理的能力。本文介绍了集成的详细步骤，并提供了一个实际案例，帮助您理解如何在实际项目中应用这些技术。

附加资源​

练习​

  1. 尝试在本地环境中安装并配置 Prometheus 和 Airflow。
  2. 创建一个简单的 DAG，并使用 Prometheus 监控其执行时间。
  3. 设置一个 Prometheus 告警规则，当任务执行时间超过 30 秒时触发告警。