使用 Google Transfer Operator 转移数据

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

注意 ：本页面尚未针对 Cloud Composer 3 进行修订，它会显示 Cloud Composer 2 的内容。

本页面演示了如何在 DAG 中使用 Google Transfer Operator 从其他服务转移数据。

Google 转移运算符简介

Google 转移运算符是一组 Airflow 运算符，可用于将数据从其他服务拉取到 Google Cloud。

本指南介绍适用于 Cloud Storage 的 Azure FileShare Storage 和 Amazon S3 运算符。有很多转移运算符可以使用 Google Cloud 中的服务以及除 Google Cloud 之外的服务。

Amazon S3 到 Cloud Storage

本部分演示如何将 Amazon S3 中的数据同步到 Cloud Storage 存储桶。

安装 Amazon 提供方软件包

apache-airflow-providers-amazon 软件包包含与 Amazon S3 进行交互的连接类型和功能。在您的环境中安装此 PyPI 软件包。

配置与 Amazon S3 的连接

Amazon 提供商软件包提供适用于 Amazon S3 的连接类型。您将创建此类型的连接。您的环境中已设置名为 google_cloud_default 的 Cloud Storage 连接。

通过以下方式设置与 Amazon S3 的连接：

1. 在 Airflow 界面中，依次前往管理 > 连接。
2. 创建新连接。
3. 选择 Amazon S3 作为连接类型。
4. 以下示例使用名为 aws_s3 的连接。您可以使用此名称，也可以为连接使用任何其他名称。
5. 按照 Amazon Web Services Connection 的 Airflow 文档中的说明指定连接参数。例如，要设置与 AWS 访问密钥的连接，您需要在 AWS 上为您的帐号生成一个访问密钥，然后提供 AWS 访问密钥 ID 作为 AWS 私有访问密钥的登录密码作为连接密码。

注意 ：我们建议您将连接的所有凭据存储在 Secret Manager 中。如需了解详情，请参阅为您的环境配置 Secret Manager。例如，您可以创建一个名为 airflow-connections-aws_s3 的 Secret，用于存储 aws_s3 连接的凭据。

从 Amazon S3 转移数据

如果您稍后想要在另一个 DAG 或任务中对同步的数据执行操作，请将其拉取到环境存储桶的 /data 文件夹中。此文件夹会同步到其他 Airflow 工作器，以便 DAG 中的任务可以对其执行操作。

以下示例 DAG 会执行以下操作：

• 将 S3 存储桶中的 /data-for-gcs 目录内容同步到环境存储桶中的 /data/from-s3/data-for-gcs/ 文件夹。
• 等待两分钟，让数据同步到您环境中的所有 Airflow 工作器。
• 使用 ls 命令输出此目录中的文件列表。请将此任务替换为适用于您数据的其他 Airflow 运算符。

import datetime
import airflow
from airflow.providers.google.cloud.transfers.s3_to_gcs import S3ToGCSOperator
from airflow.operators.bash_operator import BashOperator

with airflow.DAG(
    'composer_sample_aws_to_gcs',
    start_date=datetime.datetime(2022, 1, 1),
    schedule_interval=None,
) as dag:

    transfer_dir_from_s3 = S3ToGCSOperator(
        task_id='transfer_dir_from_s3',
        aws_conn_id='aws_s3',
        prefix='data-for-gcs',
        bucket='example-s3-bucket-transfer-operators',
        dest_gcs='gs://us-central1-example-environ-361f2312-bucket/data/from-s3/')

    sleep_2min = BashOperator(
        task_id='sleep_2min',
        bash_command='sleep 2m')

    print_dir_files = BashOperator(
        task_id='print_dir_files',
        bash_command='ls /home/airflow/gcs/data/from-s3/data-for-gcs/')


    transfer_dir_from_s3 >> sleep_2min >> print_dir_files

从 Azure 文件共享到 Cloud Storage

本部分演示如何将 Azure FileShare 中的数据同步到 Cloud Storage 存储桶。

安装 Microsoft Azure 提供方软件包

apache-airflow-providers-microsoft-azure 软件包包含与 Microsoft Azure 交互的连接类型和功能。在您的环境中安装此 PyPI 软件包。

配置与 Azure 文件共享的连接

Microsoft Azure 提供方软件包提供 Azure 文件共享的连接类型。您将创建此类连接。您的环境中已设置名为 google_cloud_default 的 Cloud Storage 连接。

通过以下方式设置与 Azure FileShare 的连接：

1. 在 Airflow 界面中，依次前往管理 > 连接。
2. 创建新连接。
3. 选择 Azure FileShare 作为连接类型。
4. 以下示例使用名为 azure_fileshare 的连接。您可以使用此名称，也可以为连接使用任何其他名称。
5. 按照 Airflow 文档中 Microsoft Azure 文件共享连接的说明指定连接参数。例如，您可以为存储帐号访问密钥指定连接字符串。

注意 ：我们建议您将连接的所有凭据存储在 Secret Manager 中。如需了解详情，请参阅为您的环境配置 Secret Manager。例如，您可以创建一个名为 airflow-connections-azure_fileshare 的 Secret，用于存储 azure_fileshare 连接的凭据。

从 Azure 文件共享传输数据

如果您稍后想要在另一个 DAG 或任务中对同步的数据执行操作，请将其拉取到环境存储桶的 /data 文件夹中。此文件夹会同步到其他 Airflow 工作器，以便 DAG 中的任务可以对其执行操作。

以下 DAG 会执行以下操作：

以下示例 DAG 会执行以下操作：

• 将 Azure 文件共享中 /data-for-gcs 目录的内容同步到环境存储桶中的 /data/from-azure 文件夹。
• 等待两分钟，让数据同步到您环境中的所有 Airflow 工作器。
• 使用 ls 命令输出此目录中的文件列表。请将此任务替换为适用于您数据的其他 Airflow 运算符。

import datetime
import airflow
from airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs import AzureFileShareToGCSOperator
from airflow.operators.bash_operator import BashOperator

with airflow.DAG(
    'composer_sample_azure_to_gcs',
    start_date=datetime.datetime(2022, 1, 1),
    schedule_interval=None,
) as dag:

    transfer_dir_from_azure = AzureFileShareToGCSOperator(
        task_id='transfer_dir_from_azure',
        azure_fileshare_conn_id='azure_fileshare',
        share_name='example-file-share',
        directory_name='data-for-gcs',
        dest_gcs='gs://us-central1-example-environ-361f2312-bucket/data/from-azure/')

    sleep_2min = BashOperator(
        task_id='sleep_2min',
        bash_command='sleep 2m')

    print_dir_files = BashOperator(
        task_id='print_dir_files',
        bash_command='ls /home/airflow/gcs/data/from-azure/')


    transfer_dir_from_azure >> sleep_2min >> print_dir_files

后续步骤

• 使用 GKE 运算符