跳到主要内容

Airflow Kubernetes 部署

Apache Airflow 是一个功能强大的工作流编排工具，广泛用于数据管道的调度和监控。随着现代基础设施的发展，Kubernetes 已成为部署和管理分布式应用程序的首选平台。本文将详细介绍如何在 Kubernetes 上部署 Airflow，并探讨其扩展和管理的最佳实践。

什么是 Airflow Kubernetes 部署？​

Airflow Kubernetes 部署是指将 Apache Airflow 的各个组件（如 Web 服务器、调度器、工作器等）部署到 Kubernetes 集群中。通过 Kubernetes，您可以轻松地扩展 Airflow 的各个组件，并利用其强大的资源管理和调度功能。

为什么选择 Kubernetes？​

  • 弹性扩展：Kubernetes 可以根据负载自动扩展或缩减 Airflow 的工作器。
  • 高可用性：Kubernetes 可以确保 Airflow 的各个组件在节点故障时自动恢复。
  • 资源隔离：每个任务都可以在独立的容器中运行，确保资源隔离和安全性。

部署步骤​

1. 准备 Kubernetes 集群​

在开始之前，您需要确保已经有一个可用的 Kubernetes 集群。如果您没有现成的集群，可以使用 Minikube 或 Kind 在本地创建一个。

bash
minikube start

2. 安装 Helm​

Helm 是 Kubernetes 的包管理工具，可以简化 Airflow 的部署过程。首先，安装 Helm：

bash
curl https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 | bash

3. 添加 Airflow Helm Chart​

Airflow 社区提供了一个官方的 Helm Chart，您可以通过以下命令添加：

bash
helm repo add apache-airflow https://airflow.apache.org
helm repo update

4. 部署 Airflow​

使用 Helm 部署 Airflow 非常简单。以下命令将在 Kubernetes 集群中部署 Airflow：

bash
helm install airflow apache-airflow/airflow --namespace airflow --create-namespace
备注

您可以通过 values.yaml 文件自定义 Airflow 的配置。例如，设置数据库连接、调整资源限制等。

5. 访问 Airflow Web UI​

部署完成后，您可以通过以下命令获取 Airflow Web UI 的访问地址：

bash
kubectl get svc -n airflow

找到 airflow-web 服务的 EXTERNAL-IP，然后在浏览器中访问该 IP 地址。

实际案例​

假设您有一个数据管道，每天需要从多个数据源提取数据，进行转换后加载到数据仓库中。通过 Airflow Kubernetes 部署，您可以轻松地扩展工作器以处理大量的并行任务，并确保任务的高可用性。

示例 DAG​

以下是一个简单的 DAG 示例，展示了如何从 S3 提取数据并加载到 PostgreSQL 数据库中：

python
from airflow import DAG
from airflow.providers.amazon.aws.operators.s3_to_sql import S3ToSqlOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
}

with DAG('s3_to_postgres', default_args=default_args, schedule_interval='@daily') as dag:
    extract = S3ToSqlOperator(
        task_id='extract_from_s3',
        s3_bucket='my-bucket',
        s3_key='data.csv',
        sql_conn_id='postgres_default',
        table='my_table',
    )

    transform = PostgresOperator(
        task_id='transform_data',
        sql='UPDATE my_table SET column = value WHERE condition;',
    )

    extract >> transform

总结​

通过 Kubernetes 部署 Airflow，您可以充分利用 Kubernetes 的弹性扩展和高可用性特性，确保数据管道的稳定运行。本文介绍了如何使用 Helm 在 Kubernetes 上部署 Airflow，并提供了一个实际案例来展示其应用场景。

附加资源​

练习​

  1. 在本地 Kubernetes 集群中部署 Airflow，并尝试运行一个简单的 DAG。
  2. 修改 values.yaml 文件，调整 Airflow 的资源限制，并观察其对性能的影响。
  3. 尝试使用 Kubernetes Executor 来运行任务，并比较其与 Celery Executor 的差异。