跳到主要内容

Airflow 与AWS集成

Apache Airflow 是一个开源的工作流自动化工具，广泛用于调度和监控复杂的数据管道。AWS（Amazon Web Services）是全球领先的云服务平台，提供了丰富的计算、存储和分析服务。将 Airflow 与 AWS 集成，可以帮助您自动化和管理云上的工作流，从而提高效率和可扩展性。

为什么需要将 Airflow 与 AWS 集成？​

Airflow 的核心功能是调度和监控任务，而 AWS 提供了强大的云基础设施。通过将两者集成，您可以：

  • 自动化 AWS 服务的任务调度，如启动 EC2 实例、运行 EMR 作业或触发 Lambda 函数。
  • 监控和管理云上的工作流，确保任务按计划执行。
  • 利用 AWS 的弹性计算资源，动态扩展 Airflow 的工作负载。

准备工作​

在开始之前，请确保您已经完成以下准备工作：

  1. 安装 Airflow：您可以在本地或云服务器上安装 Airflow。推荐使用 Docker 进行快速部署。
  2. 创建 AWS 账户：如果您还没有 AWS 账户，请先注册并创建一个账户。
  3. 配置 AWS CLI：在本地或 Airflow 服务器上安装并配置 AWS CLI，以便与 AWS 服务进行交互。

配置 Airflow 与 AWS 的连接​

要将 Airflow 与 AWS 集成，首先需要在 Airflow 中配置 AWS 连接。以下是具体步骤：

  1. 安装 AWS Provider 包：Airflow 提供了 apache-airflow-providers-amazon 包，用于与 AWS 服务集成。您可以通过以下命令安装：

    pip install apache-airflow-providers-amazon

  2. 配置 AWS 连接：在 Airflow 的 Web UI 中，导航到 Admin > Connections，然后点击 Add a new record。填写以下信息：

    • Conn Id: aws_default
    • Conn Type: Amazon Web Services
    • Login: 您的 AWS Access Key ID
    • Password: 您的 AWS Secret Access Key
    • Extra: 可选，可以指定 AWS 区域，例如 {"region_name": "us-west-2"}
    提示

    为了安全起见，建议使用 IAM 角色或临时凭证，而不是直接在连接中存储 AWS 凭证。

使用 Airflow 操作 AWS 服务​

配置好连接后，您可以在 Airflow DAG 中使用 AWS 服务。以下是一些常见的用例：

1. 启动 EC2 实例​

以下是一个简单的 DAG 示例，用于启动一个 EC2 实例：

from airflow import DAG
from airflow.providers.amazon.aws.operators.ec2 import EC2StartInstanceOperator
from airflow.utils.dates import days_ago

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

with DAG('start_ec2_instance', default_args=default_args, schedule_interval=None) as dag:
    start_instance = EC2StartInstanceOperator(
        task_id='start_ec2_instance',
        instance_id='i-0abcdef1234567890',
        aws_conn_id='aws_default'
    )

2. 运行 EMR 作业​

以下是一个 DAG 示例，用于在 AWS EMR 上运行一个 Spark 作业：

from airflow import DAG
from airflow.providers.amazon.aws.operators.emr import EmrCreateJobFlowOperator, EmrAddStepsOperator
from airflow.utils.dates import days_ago

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

with DAG('run_emr_job', default_args=default_args, schedule_interval=None) as dag:
    create_job_flow = EmrCreateJobFlowOperator(
        task_id='create_job_flow',
        job_flow_overrides={
            'Name': 'My Spark Job',
            'ReleaseLabel': 'emr-6.2.0',
            'Applications': [{'Name': 'Spark'}],
            'Instances': {
                'InstanceGroups': [
                    {
                        'Name': 'Master nodes',
                        'Market': 'ON_DEMAND',
                        'InstanceRole': 'MASTER',
                        'InstanceType': 'm5.xlarge',
                        'InstanceCount': 1,
                    },
                    {
                        'Name': 'Worker nodes',
                        'Market': 'ON_DEMAND',
                        'InstanceRole': 'CORE',
                        'InstanceType': 'm5.xlarge',
                        'InstanceCount': 2,
                    },
                ],
                'KeepJobFlowAliveWhenNoSteps': False,
                'TerminationProtected': False,
            },
            'Steps': [],
            'VisibleToAllUsers': True,
            'JobFlowRole': 'EMR_EC2_DefaultRole',
            'ServiceRole': 'EMR_DefaultRole',
        },
        aws_conn_id='aws_default'
    )

    add_steps = EmrAddStepsOperator(
        task_id='add_steps',
        job_flow_id="{{ task_instance.xcom_pull(task_ids='create_job_flow', key='return_value') }}",
        steps=[
            {
                'Name': 'Run Spark Job',
                'ActionOnFailure': 'CONTINUE',
                'HadoopJarStep': {
                    'Jar': 'command-runner.jar',
                    'Args': ['spark-submit', 's3://my-bucket/my-spark-job.py'],
                },
            }
        ],
        aws_conn_id='aws_default'
    )

    create_job_flow >> add_steps

实际案例：自动化数据管道​

假设您有一个数据管道，需要每天从 S3 读取数据，进行处理，然后将结果存储到 Redshift 中。您可以使用 Airflow 来自动化这个流程：

  1. 从 S3 读取数据：使用 S3ToRedshiftOperator 将数据从 S3 加载到 Redshift。
  2. 处理数据：使用 PythonOperator 或 SparkSubmitOperator 处理数据。
  3. 将结果存储到 Redshift：再次使用 S3ToRedshiftOperator 将处理后的数据存储到 Redshift。

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

from airflow import DAG
from airflow.providers.amazon.aws.transfers.s3_to_redshift import S3ToRedshiftOperator
from airflow.operators.python_operator import PythonOperator
from airflow.utils.dates import days_ago

def process_data():
    # 这里可以添加数据处理逻辑
    pass

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

with DAG('data_pipeline', default_args=default_args, schedule_interval='@daily') as dag:
    load_data = S3ToRedshiftOperator(
        task_id='load_data',
        schema='my_schema',
        table='my_table',
        s3_bucket='my-bucket',
        s3_key='data/raw/',
        redshift_conn_id='redshift_default',
        aws_conn_id='aws_default'
    )

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

    store_results = S3ToRedshiftOperator(
        task_id='store_results',
        schema='my_schema',
        table='my_table_processed',
        s3_bucket='my-bucket',
        s3_key='data/processed/',
        redshift_conn_id='redshift_default',
        aws_conn_id='aws_default'
    )

    load_data >> process_data_task >> store_results

总结​

通过将 Airflow 与 AWS 集成，您可以轻松地自动化和管理云上的工作流。本文介绍了如何配置 Airflow 与 AWS 的连接，并提供了几个常见的用例和实际案例。希望这些内容能帮助您更好地理解和使用 Airflow 与 AWS 的集成。

附加资源与练习​

  • AWS 官方文档：了解更多关于 AWS 服务的使用方法。
  • Airflow 官方文档：深入理解 Airflow 的功能和配置。
  • 练习：尝试创建一个 DAG，自动化一个包含多个 AWS 服务的复杂工作流，如从 S3 读取数据、在 EMR 上处理数据，并将结果存储到 Redshift 中。
备注

如果您在集成过程中遇到问题，可以参考 Airflow 和 AWS 的官方文档，或者在社区论坛中寻求帮助。