Déclencher des DAG Cloud Composer avec des fonctions Cloud Run et l'API REST Airflow

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Cette page explique comment utiliser les fonctions Cloud Run pour déclencher Graphes orientés acycliques (DAG) Cloud Composer en réponse à des événements

Apache Airflow est conçu pour exécuter des DAG de façon régulière, mais vous pouvez aussi déclencher des DAG en réponse à des événements. Pour ce faire, vous pouvez utiliser des fonctions Cloud Run afin de déclencher des DAG Cloud Composer lorsqu'un événement spécifique se produit.

L'exemple de ce guide exécute un DAG à chaque modification au sein d'un bucket Cloud Storage. Les modifications apportées à tout objet d'un bucket déclenchent une fonction. Cette fonction envoie une requête à l'API REST Airflow de votre environnement Cloud Composer. Airflow traite cette requête et exécute un DAG. Le DAG fournit des informations sur la modification.

Avant de commencer

Vérifier la configuration réseau de votre environnement

Cette solution ne fonctionne pas dans les configurations d'adresses IP privées et VPC Service Controls car il n'est pas possible de configurer la connectivité à partir des fonctions Cloud Run au serveur Web Airflow dans ces configurations.

Dans Cloud Composer 2, vous pouvez utiliser une autre approche: Déclencher des DAG à l'aide des fonctions Cloud Run et des messages Pub/Sub

Activer les API pour votre projet.

Console

Enable the Cloud Composer and Cloud Run functions APIs.

Enable the APIs

gcloud

Enable the Cloud Composer and Cloud Run functions APIs: 

gcloud services enable cloudfunctions.googleapis.com composer.googleapis.com

Activer l'API REST Airflow

Pour Airflow 2, l'API REST stable est déjà activée par défaut. Si l'API stable est désactivée dans votre environnement, activez l'API REST stable.

Autoriser les appels d'API à l'API REST Airflow à l'aide du contrôle d'accès du serveur Web

Les fonctions Cloud Run peuvent contacter l'API REST Airflow via IPv4 ou IPv6.

Si vous n'êtes pas sûr de la plage d'adresses IP à l'origine de l'appel, utilisez une plage d'adresses IP par défaut l'option de configuration All IP addresses have access (default) dans le contrôle d'accès du serveur Web, pour ne pas bloquer accidentellement vos fonctions Cloud Run.

Créer un bucket Cloud Storage

Étant donné que cet exemple déclenche un DAG en réponse aux modifications apportées à un bucket Cloud Storage, créez un bucket à utiliser dans cet exemple.

Obtenir l'URL du serveur Web Airflow

Cet exemple envoie des requêtes API REST au point de terminaison du serveur Web Airflow. Vous utilisez l'URL du serveur Web Airflow dans le code de votre fonction Cloud.

Console

  1. Dans la console Google Cloud, accédez à la page Environnements.

    Accéder à la page Environnements

  2. Cliquez sur le nom de votre environnement.

  3. Sur la page Détails de l'environnement, accédez à l'onglet Configuration de l'environnement.

  4. L'URL du serveur Web Airflow est répertoriée dans l'élément UI Web Airflow.

gcloud

Exécutez la commande suivante :

gcloud composer environments describe ENVIRONMENT_NAME \
    --location LOCATION \
    --format='value(config.airflowUri)'

Remplacez :

  • ENVIRONMENT_NAME par le nom de l'environnement.
  • LOCATION par la région où se trouve l'environnement.

Importer un DAG dans votre environnement

Importer un DAG dans votre environnement. L'exemple de DAG suivant génère la configuration d'exécution de DAG reçue. Vous allez déclencher ce DAG à partir d'une fonction que vous créerez plus loin dans ce guide.

import datetime

import airflow
from airflow.operators.bash import BashOperator


with airflow.DAG(
    "composer_sample_trigger_response_dag",
    start_date=datetime.datetime(2021, 1, 1),
    # Not scheduled, trigger only
    schedule_interval=None,
) as dag:
    # Print the dag_run's configuration, which includes information about the
    # Cloud Storage object change.
    print_gcs_info = BashOperator(
        task_id="print_gcs_info", bash_command="echo {{ dag_run.conf }}"
    )

Déployer une fonction Cloud qui déclenche le DAG

Vous pouvez déployer une fonction Cloud à l'aide de votre langage préféré compatible avec les fonctions Cloud Run ou Cloud Run. Ce tutoriel présente Fonction Cloud implémentée en Python Java

Spécifier les paramètres de configuration de la fonction Cloud

  • Déclencheur. Pour cet exemple, sélectionnez un déclencheur qui s'active lorsqu'un objet est créé dans un bucket ou qu'un objet existant est écrasé.

    • Type de déclencheur. Cloud Storage.

    • Type d'événement. Finaliser/Créer.

    • Bucket. Sélectionnez un bucket qui doit déclencher cette fonction.

    • Réessayer après échec. Nous vous recommandons de désactiver cette option pour les besoins de cet exemple. Si vous utilisez votre propre fonction dans un environnement de production, activez cette option pour gérer les erreurs temporaires.

  • Compte de service d'exécution, dans le Section Paramètres d'exécution, de compilation, de connexion et de sécurité Utilisez l'une des options suivantes en fonction de vos préférences :

    • Sélectionnez le compte de service Compute Engine par défaut. Avec les autorisations IAM par défaut, ce compte peut exécuter des fonctions qui accèdent aux environnements Cloud Composer.

    • Créez un compte de service personnalisé doté du rôle Utilisateur de Composer et spécifiez-le en tant que compte de service d'exécution pour cette fonction. Cette option adhère au principe du moindre privilège.

  • Environnement d'exécution et point d'entrée de l'étape Code :

    • (Python) Lorsque vous ajoutez du code pour cet exemple, sélectionnez l'environnement d'exécution Python 3.7 ou version ultérieure, puis spécifiez trigger_dag_gcf comme point d'entrée.

    • (Java) Lors de l'ajout du code pour cet exemple, sélectionnez Java 17. et spécifiez com.example.Example comme point d'entrée.

Ajouter des conditions

Python

Spécifiez les dépendances dans le fichier requirements.txt :

google-auth==2.19.1
requests==2.32.2

Java

Ajoutez les dépendances suivantes à la section dependencies dans le fichier pom.xml généré par l'interface utilisateur de Google Cloud Functions.

    <dependency>
      <groupId>com.google.apis</groupId>
      <artifactId>google-api-services-docs</artifactId>
      <version>v1-rev20210707-1.32.1</version>
    </dependency>
    <dependency>
      <groupId>com.google.api-client</groupId>
      <artifactId>google-api-client</artifactId>
      <version>1.32.1</version>
    </dependency>
    <dependency>
      <groupId>com.google.auth</groupId>
      <artifactId>google-auth-library-credentials</artifactId>
      <version>1.14.0</version>
    </dependency>
    <dependency>
      <groupId>com.google.auth</groupId>
      <artifactId>google-auth-library-oauth2-http</artifactId>
      <version>1.14.0</version>
    </dependency>

Python

Ajoutez le code permettant de déclencher les DAG à l'aide de l'API REST Airflow. Créer un fichier nommé composer2_airflow_rest_api.py et insérer le code permettant de créer l'API REST Airflow dans ce fichier.

Ne modifiez aucune variable. La fonction Cloud importe ce fichier à partir du fichier main.py.

from __future__ import annotations

from typing import Any

import google.auth
from google.auth.transport.requests import AuthorizedSession
import requests


# Following GCP best practices, these credentials should be
# constructed at start-up time and used throughout
# https://cloud.google.com/apis/docs/client-libraries-best-practices
AUTH_SCOPE = "https://www.googleapis.com/auth/cloud-platform"
CREDENTIALS, _ = google.auth.default(scopes=[AUTH_SCOPE])


def make_composer2_web_server_request(
    url: str, method: str = "GET", **kwargs: Any
) -> google.auth.transport.Response:
    """
    Make a request to Cloud Composer 2 environment's web server.
    Args:
      url: The URL to fetch.
      method: The request method to use ('GET', 'OPTIONS', 'HEAD', 'POST', 'PUT',
        'PATCH', 'DELETE')
      **kwargs: Any of the parameters defined for the request function:
                https://github.com/requests/requests/blob/master/requests/api.py
                  If no timeout is provided, it is set to 90 by default.
    """

    authed_session = AuthorizedSession(CREDENTIALS)

    # Set the default timeout, if missing
    if "timeout" not in kwargs:
        kwargs["timeout"] = 90

    return authed_session.request(method, url, **kwargs)


def trigger_dag(web_server_url: str, dag_id: str, data: dict) -> str:
    """
    Make a request to trigger a dag using the stable Airflow 2 REST API.
    https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html

    Args:
      web_server_url: The URL of the Airflow 2 web server.
      dag_id: The DAG ID.
      data: Additional configuration parameters for the DAG run (json).
    """

    endpoint = f"api/v1/dags/{dag_id}/dagRuns"
    request_url = f"{web_server_url}/{endpoint}"
    json_data = {"conf": data}

    response = make_composer2_web_server_request(
        request_url, method="POST", json=json_data
    )

    if response.status_code == 403:
        raise requests.HTTPError(
            "You do not have a permission to perform this operation. "
            "Check Airflow RBAC roles for your account."
            f"{response.headers} / {response.text}"
        )
    elif response.status_code != 200:
        response.raise_for_status()
    else:
        return response.text

Placez le code suivant dans le fichier main.py. Remplacez la valeur de la variable web_server_url par l'adresse du serveur Web Airflow obtenue précédemment.

# Copyright 2021 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
Trigger a DAG in a Cloud Composer 2 environment in response to an event,
using Cloud Functions.
"""

from typing import Any

import composer2_airflow_rest_api

def trigger_dag_gcf(data, context=None):
    """
    Trigger a DAG and pass event data.

    Args:
      data: A dictionary containing the data for the event. Its format depends
      on the event.
      context: The context object for the event.

    For more information about the arguments, see:
    https://cloud.google.com/functions/docs/writing/background#function_parameters
    """

    # TODO(developer): replace with your values
    # Replace web_server_url with the Airflow web server address. To obtain this
    # URL, run the following command for your environment:
    # gcloud composer environments describe example-environment \
    #  --location=your-composer-region \
    #  --format="value(config.airflowUri)"
    web_server_url = (
        "https://example-airflow-ui-url-dot-us-central1.composer.googleusercontent.com"
    )
    # Replace with the ID of the DAG that you want to run.
    dag_id = 'composer_sample_trigger_response_dag'

    composer2_airflow_rest_api.trigger_dag(web_server_url, dag_id, data)

Java

Placez le code suivant dans le fichier Example.java. Remplacez la valeur de la variable webServerUrl par l'adresse du serveur Web Airflow obtenue précédemment.


// Copyright 2022 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package com.example;

import com.example.Example.GcsEvent;
import com.google.api.client.http.GenericUrl;
import com.google.api.client.http.HttpContent;
import com.google.api.client.http.HttpRequest;
import com.google.api.client.http.HttpRequestFactory;
import com.google.api.client.http.HttpResponse;
import com.google.api.client.http.HttpResponseException;
import com.google.api.client.http.javanet.NetHttpTransport;
import com.google.api.client.http.json.JsonHttpContent;
import com.google.api.client.json.gson.GsonFactory;
import com.google.auth.http.HttpCredentialsAdapter;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.cloud.functions.BackgroundFunction;
import com.google.cloud.functions.Context;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Logger;

/**
 * Cloud Function that triggers an Airflow DAG in response to an event (in
 * this case a Cloud Storage event).
 */
public class Example implements BackgroundFunction<GcsEvent> {
  private static final Logger logger = Logger.getLogger(Example.class.getName());

  // TODO(developer): replace with your values
  // Replace webServerUrl with the Airflow web server address. To obtain this
  // URL, run the following command for your environment:
  // gcloud composer environments describe example-environment \
  //  --location=your-composer-region \
  //  --format="value(config.airflowUri)"
  @Override
  public void accept(GcsEvent event, Context context) throws Exception {
    String webServerUrl = "https://example-airflow-ui-url-dot-us-central1.composer.googleusercontent.com";
    String dagName = "composer_sample_trigger_response_dag";
    String url = String.format("%s/api/v1/dags/%s/dagRuns", webServerUrl, dagName);

    logger.info(String.format("Triggering DAG %s as a result of an event on the object %s.",
      dagName, event.name));
    logger.info(String.format("Triggering DAG via the following URL: %s", url));

    GoogleCredentials googleCredentials = GoogleCredentials.getApplicationDefault()
        .createScoped("https://www.googleapis.com/auth/cloud-platform");
    HttpCredentialsAdapter credentialsAdapter = new HttpCredentialsAdapter(googleCredentials);
    HttpRequestFactory requestFactory =
      new NetHttpTransport().createRequestFactory(credentialsAdapter);

    Map<String, Map<String, String>> json = new HashMap<String, Map<String, String>>();
    Map<String, String> conf = new HashMap<String, String>();
    conf.put("bucket", event.bucket);
    conf.put("name", event.name);
    conf.put("generation", event.generation);
    conf.put("operation", context.eventType());
    json.put("conf", conf);
    HttpContent content = new JsonHttpContent(new GsonFactory(), json);
    HttpRequest request = requestFactory.buildPostRequest(new GenericUrl(url), content);
    request.getHeaders().setContentType("application/json");
    HttpResponse response;
    try {
      response = request.execute();
      int statusCode = response.getStatusCode();
      logger.info("Response code: " + statusCode);
      logger.info(response.parseAsString());
    } catch (HttpResponseException e) {
      // https://cloud.google.com/java/docs/reference/google-http-client/latest/com.google.api.client.http.HttpResponseException
      logger.info("Received HTTP exception");
      logger.info(e.getLocalizedMessage());
      logger.info("- 400 error: wrong arguments passed to Airflow API");
      logger.info("- 401 error: check if service account has Composer User role");
      logger.info("- 403 error: check Airflow RBAC roles assigned to service account");
      logger.info("- 404 error: check Web Server URL");
    } catch (Exception e) {
      logger.info("Received exception");
      logger.info(e.getLocalizedMessage());
    }
  }

  /** Details of the storage event. */
  public static class GcsEvent {
    /** Bucket name. */
    String bucket;
    /** Object name. */
    String name;
    /** Object version. */
    String generation;
  }
}

Tester votre fonction

Pour vérifier que votre fonction et votre DAG fonctionnent comme prévu, procédez comme suit :

  1. Attendez que votre fonction soit déployée.
  2. Importer un fichier dans votre bucket Cloud Storage Vous pouvez également déclencher la fonction manuellement en sélectionnant l'action Tester la fonction dans la console Google Cloud.
  3. Consultez la page du DAG dans l'interface Web Airflow. Le DAG doit comporter une exécution de DAG active ou déjà terminée.
  4. Dans l'UI d'Airflow, consultez les journaux des tâches associées à cette exécution. Vous devriez voir que la tâche print_gcs_info écrit les données reçues de la fonction dans les journaux :

Python 

[2021-04-04 18:25:44,778] {bash_operator.py:154} INFO - Output:
[2021-04-04 18:25:44,781] {bash_operator.py:158} INFO - Triggered from GCF:
    {bucket: example-storage-for-gcf-triggers, contentType: text/plain,
    crc32c: dldNmg==, etag: COW+26Sb5e8CEAE=, generation: 1617560727904101,
    ... }
[2021-04-04 18:25:44,781] {bash_operator.py:162} INFO - Command exited with
    return code 0h

Java 

[2023-02-08, 08:00:09 UTC] {subprocess.py:86} INFO - Output:
[2023-02-08, 08:00:09 UTC] {subprocess.py:93} INFO - {bucket: example-storage-for-gcf-triggers, generation: 1675843189006715, name: file.txt, operation: google.storage.object.create}
[2023-02-08, 08:00:09 UTC] {subprocess.py:97} INFO - Command exited with return code 0

Étape suivante