Transférer des données provenant d'autres services avec les opérateurs de transfert Google

Cloud Composer 1 | Cloud Composer 2

Cette page explique comment transférer des données à partir d'autres services à l'aide d'opérateurs de transfert Google dans vos DAG.

À propos des opérateurs de transfert Google

Les opérateurs de transfert Google sont un ensemble d'opérateurs Airflow que vous pouvez utiliser pour extraire des données d'autres services vers Google Cloud.

Ce guide présente les opérateurs pour Azure FileShare Storage et Amazon S3 qui fonctionnent avec Cloud Storage. De nombreux autres opérateurs de transfert fonctionnent avec des services dans Google Cloud et avec des services autres que Google Cloud.

Amazon S3 vers Cloud Storage

Cette section explique comment synchroniser des données d'Amazon S3 vers un bucket Cloud Storage.

Installer le package du fournisseur Amazon

Le package apache-airflow-providers-amazon contient les types de connexions et les fonctionnalités qui interagissent avec Amazon S3. Installez ce package PyPI dans votre environnement.

Configurer une connexion à Amazon S3

Le package du fournisseur Amazon fournit un type de connexion pour Amazon S3. Vous créez une connexion de ce type. La connexion pour Cloud Storage, nommée google_cloud_default, est déjà configurée dans votre environnement.

Configurez une connexion à Amazon S3 de la manière suivante:

1. Dans l'interface utilisateur Airflow, accédez à Administration > Connexions.
2. Créez une connexion.
3. Sélectionnez Amazon S3 comme type de connexion.
4. L'exemple suivant utilise une connexion nommée aws_s3. Vous pouvez utiliser ce nom ou tout autre nom pour la connexion.
5. Spécifiez les paramètres de connexion comme décrit dans la documentation Airflow pour Amazon Web Services Connection. Par exemple, pour configurer une connexion avec des clés d'accès AWS, vous devez générer une clé d'accès pour votre compte sur AWS, puis fournir l'ID de clé d'accès AWS sous la forme d'une clé d'accès secrète AWS et servir de mot de passe pour la connexion.

Remarque :Nous vous recommandons de stocker tous les identifiants pour les connexions dans Secret Manager. Pour en savoir plus, consultez la section Configurer Secret Manager pour votre environnement. Par exemple, vous pouvez créer un secret nommé airflow-connections-aws_s3 qui stocke les identifiants de la connexion aws_s3.

Transférer des données depuis Amazon S3

Si vous souhaitez effectuer des opérations sur les données synchronisées ultérieurement dans un autre DAG ou une autre tâche, extrayez-les dans le dossier /data du bucket de votre environnement. Ce dossier est synchronisé avec d'autres nœuds de calcul Airflow, de sorte que les tâches de votre DAG puissent y être exécutées.

L'exemple de DAG suivant effectue les opérations suivantes:

• Synchronise le contenu du répertoire /data-for-gcs depuis un bucket S3 vers le dossier /data/from-s3/data-for-gcs/ du bucket de votre environnement.
• Attend deux minutes que les données se synchronisent avec tous les nœuds de calcul Airflow de votre environnement.
• Génère la liste des fichiers de ce répertoire à l'aide de la commande ls. Remplacez cette tâche par d'autres opérateurs Airflow qui utilisent vos données.

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 FileShare vers Cloud Storage

Cette section explique comment synchroniser des données d'Azure FileShare avec un bucket Cloud Storage.

Installer le package du fournisseur Microsoft Azure

Le package apache-airflow-providers-microsoft-azure contient les types de connexion et les fonctionnalités qui interagissent avec Microsoft Azure. Installez ce package PyPI dans votre environnement.

Configurer une connexion à Azure FileShare

Le package du fournisseur Microsoft Azure fournit un type de connexion pour Azure File Share. Vous créez une connexion de ce type. La connexion pour Cloud Storage, nommée google_cloud_default, est déjà configurée dans votre environnement.

Configurez une connexion à Azure FileShare de la manière suivante:

1. Dans l'interface utilisateur Airflow, accédez à Administration > Connexions.
2. Créez une connexion.
3. Sélectionnez Azure FileShare comme type de connexion.
4. L'exemple suivant utilise une connexion nommée azure_fileshare. Vous pouvez utiliser ce nom ou tout autre nom pour la connexion.
5. Spécifiez les paramètres de connexion comme décrit dans la documentation Airflow pour la connexion au partage de fichiers Microsoft Azure. Par exemple, vous pouvez spécifier une chaîne de connexion pour votre clé d'accès au compte de stockage.

Remarque :Nous vous recommandons de stocker tous les identifiants pour les connexions dans Secret Manager. Pour en savoir plus, consultez la section Configurer Secret Manager pour votre environnement. Par exemple, vous pouvez créer un secret nommé airflow-connections-azure_fileshare qui stocke les identifiants de la connexion azure_fileshare.

Transférer des données depuis Azure FileShare

Si vous souhaitez effectuer des opérations sur les données synchronisées ultérieurement dans un autre DAG ou une autre tâche, extrayez-les dans le dossier /data du bucket de votre environnement. Ce dossier est synchronisé avec d'autres nœuds de calcul Airflow, de sorte que les tâches de votre DAG puissent y être exécutées.

Le DAG suivant effectue les opérations suivantes:

L'exemple de DAG suivant effectue les opérations suivantes:

• Synchronise le contenu du répertoire /data-for-gcs depuis Azure File Share vers le dossier /data/from-azure du bucket de votre environnement.
• Attend deux minutes que les données se synchronisent avec tous les nœuds de calcul Airflow de votre environnement.
• Génère la liste des fichiers de ce répertoire à l'aide de la commande ls. Remplacez cette tâche par d'autres opérateurs Airflow qui utilisent vos données.

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

Étapes suivantes

• Utiliser des opérateurs GKE