Acionamento de DAGs com o Cloud Functions

Mantenha tudo organizado com as coleções Salve e categorize o conteúdo com base nas suas preferências.

Cloud Composer 1 | Cloud Composer 2

Nesta página, veja como usar o Cloud Functions para acionar DAGs em resposta a eventos.

O Airflow foi criado para executar DAGs regularmente, mas também é possível acionar DAGs em resposta a eventos. Uma forma de fazer isso é usando o Cloud Functions para acionar DAGs do Cloud Composer quando ocorre um evento especificado. Por exemplo, é possível criar uma função que aciona um DAG quando um objeto é alterado em um bucket do Cloud Storage ou quando uma mensagem é enviada para um tópico do Pub/Sub.

No exemplo deste guia, um DAG é executado sempre que ocorre uma mudança em um bucket do Cloud Storage. Alterações em qualquer objeto em um bucket acionam uma função. Essa função faz uma solicitação à API REST do Airflow do ambiente do Cloud Composer. O Airflow processa essa solicitação e executa um DAG. O DAG mostra informações sobre a alteração.

Antes de começar

Ativar as APIs do projeto

Console

  • Ative as APIs Cloud Composer and Cloud Functions.

    Ative as APIs

  • gcloud

  • Ative as APIs Cloud Composer and Cloud Functions:

    gcloud services enable cloudfunctions.googleapis.comcomposer.googleapis.com
  • Ativar a API REST do Airflow

    Dependendo da sua versão do Airflow, faça o seguinte:

    crie um bucket do Cloud Storage

    Este exemplo aciona um DAG em resposta a alterações em um bucket do Cloud Storage. Crie um novo bucket para usar neste exemplo.

    Ver o URL do servidor da Web do Airflow

    Este exemplo faz solicitações da API REST para o endpoint do servidor da Web do Airflow. Use a parte do URL da interface da Web do Airflow antes de .appspot.com no código da função do Cloud.

    Console

    1. No Console do Google Cloud, acesse a página Ambientes.

      Acessar "Ambientes"

    2. Clique no nome do seu ambiente.

    3. Na página Detalhes do ambiente, acesse a guia Detalhes do ambiente.

    4. O URL do servidor da Web do Airflow está listado no item da IU da Web do Airflow.

    gcloud

    Execute este comando:

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

    Substitua:

    • ENVIRONMENT_NAME pelo nome do ambiente
    • LOCATION pela região em que o ambiente está localizado;

    Acessar o client_id do proxy do IAM

    Para fazer uma solicitação ao endpoint da API REST do Airflow, a função exige o ID do cliente do proxy do IAM que protege o servidor da Web do Airflow.

    O Cloud Composer não fornece essas informações diretamente. Em vez disso, faça uma solicitação não autenticada no servidor da Web do Airflow e capture o ID do cliente do URL de redirecionamento:

    cURL

    curl -v AIRFLOW_URL 2>&1 >/dev/null | grep -o "client_id\=[A-Za-z0-9-]*\.apps\.googleusercontent\.com"
    

    Substitua AIRFLOW_URL pelo URL da interface da Web do Airflow.

    Na saída, procure a string depois de client_id. Exemplo:

    client_id=836436932391-16q2c5f5dcsfnel77va9bvf4j280t35c.apps.googleusercontent.com
    

    Python

    Salve o código a seguir em um arquivo chamado get_client_id.py. Preencha os valores de project_id, location e composer_environment. Em seguida, execute o código no Cloud Shell ou no ambiente local.

    # This script is intended to be used with Composer 1 environments
    # In Composer 2, the Airflow Webserver is not in the tenant project
    # so there is no tenant client ID
    # See https://cloud.google.com/composer/docs/composer-2/environment-architecture
    # for more details
    import google.auth
    import google.auth.transport.requests
    import requests
    import six.moves.urllib.parse
    
    # Authenticate with Google Cloud.
    # See: https://cloud.google.com/docs/authentication/getting-started
    credentials, _ = google.auth.default(
        scopes=["https://www.googleapis.com/auth/cloud-platform"]
    )
    authed_session = google.auth.transport.requests.AuthorizedSession(credentials)
    
    # project_id = 'YOUR_PROJECT_ID'
    # location = 'us-central1'
    # composer_environment = 'YOUR_COMPOSER_ENVIRONMENT_NAME'
    
    environment_url = (
        "https://composer.googleapis.com/v1beta1/projects/{}/locations/{}"
        "/environments/{}"
    ).format(project_id, location, composer_environment)
    composer_response = authed_session.request("GET", environment_url)
    environment_data = composer_response.json()
    composer_version = environment_data["config"]["softwareConfig"]["imageVersion"]
    if "composer-1" not in composer_version:
        version_error = ("This script is intended to be used with Composer 1 environments. "
                         "In Composer 2, the Airflow Webserver is not in the tenant project, "
                         "so there is no tenant client ID. "
                         "See https://cloud.google.com/composer/docs/composer-2/environment-architecture for more details.")
        raise (RuntimeError(version_error))
    airflow_uri = environment_data["config"]["airflowUri"]
    
    # The Composer environment response does not include the IAP client ID.
    # Make a second, unauthenticated HTTP request to the web server to get the
    # redirect URI.
    redirect_response = requests.get(airflow_uri, allow_redirects=False)
    redirect_location = redirect_response.headers["location"]
    
    # Extract the client_id query parameter from the redirect.
    parsed = six.moves.urllib.parse.urlparse(redirect_location)
    query_string = six.moves.urllib.parse.parse_qs(parsed.query)
    print(query_string["client_id"][0])

    Como acionar um DAG a partir do Cloud Functions

    Fazer upload de um DAG para o ambiente

    Faça o upload de um DAG para seu ambiente. O exemplo a seguir mostra a configuração de execução do DAG recebida. Você acionará esse DAG a partir de uma função que vai criar neste guia depois.

    import datetime
    
    import airflow
    from airflow.operators.bash_operator 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 }}')

    Implantar uma Função do Cloud que aciona o DAG

    Implante um Cloud Function no Python usando os parâmetros e o conteúdo de configuração a seguir.

    Especificar os parâmetros de configuração da função do Cloud

    • Gatilho. Neste exemplo, selecione um gatilho que funcione quando um novo objeto for criado em um bucket ou um objeto existente for substituído.

      • Tipo de gatilho. Cloud Storage.

      • Tipo de evento. Finalizar/Criar.

      • Bucket. Selecione um bucket que precisa acionar essa função.

      • Tentar novamente em caso de falha. Recomendamos desativar essa opção para os fins deste exemplo. Se você usar sua própria função em um ambiente de produção, ative essa opção para processar erros temporários.

    • Conta de serviço do ambiente de execução, na seção Configurações de ambiente de execução, build, conexões e segurança. Use uma das seguintes opções, dependendo das suas preferências:

      • Selecione Conta de serviço padrão do Compute Engine. Com as permissões padrão do IAM, essa conta pode executar funções que acessam os ambientes do Cloud Composer.

      • Crie uma conta de serviço personalizada que tenha o papel de Usuário do Composer e especifique-a como uma conta de serviço do ambiente de execução para essa função. Essa opção segue o princípio do privilégio mínimo.

    • Ambiente de execução e ponto de entrada, na etapa Código. Ao adicionar o código deste exemplo, selecione o ambiente de execução Python 3.7 ou posterior e especifique trigger_dag como o ponto de entrada.

    Adicionar requisitos

    Especifique as dependências no arquivo requirements.txt:

    requests_toolbelt==0.9.1
    google-auth==2.6.2
    

    Adicionar o código da função do Cloud

    Coloque o código a seguir no arquivo main.py e faça as seguintes substituições:

    • Substitua o valor da variável client_id pelo valor client_id recebido anteriormente.

    • Substitua o valor da variável webserver_id pelo ID do projeto de locatário, que faz parte do URL da interface da Web do Airflow antes de .appspot.com. Você já recebeu o URL da interface da Web do Airflow.

    • Especifique a versão da API REST do Airflow usada:

      • Se você usa a API REST estável do Airflow, defina a variável USE_EXPERIMENTAL_API como False.
      • Se você usa a API REST experimental do Airflow, não é necessário fazer alterações. A variável USE_EXPERIMENTAL_API já está definida como True.
    
    from google.auth.transport.requests import Request
    from google.oauth2 import id_token
    import requests
    
    IAM_SCOPE = 'https://www.googleapis.com/auth/iam'
    OAUTH_TOKEN_URI = 'https://www.googleapis.com/oauth2/v4/token'
    # If you are using the stable API, set this value to False
    # For more info about Airflow APIs see https://cloud.google.com/composer/docs/access-airflow-api
    USE_EXPERIMENTAL_API = True
    
    def trigger_dag(data, context=None):
        """Makes a POST request to the Composer DAG Trigger API
    
        When called via Google Cloud Functions (GCF),
        data and context are Background function parameters.
    
        For more info, refer to
        https://cloud.google.com/functions/docs/writing/background#functions_background_parameters-python
    
        To call this function from a Python script, omit the ``context`` argument
        and pass in a non-null value for the ``data`` argument.
    
        This function is currently only compatible with Composer v1 environments.
        """
    
        # Fill in with your Composer info here
        # Navigate to your webserver's login page and get this from the URL
        # Or use the script found at
        # https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/composer/rest/get_client_id.py
        client_id = 'YOUR-CLIENT-ID'
        # This should be part of your webserver's URL:
        # {tenant-project-id}.appspot.com
        webserver_id = 'YOUR-TENANT-PROJECT'
        # The name of the DAG you wish to trigger
        dag_name = 'composer_sample_trigger_response_dag'
    
        if USE_EXPERIMENTAL_API:
            endpoint = f'api/experimental/dags/{dag_name}/dag_runs'
            json_data = {'conf': data, 'replace_microseconds': 'false'}
        else:
            endpoint = f'api/v1/dags/{dag_name}/dagRuns'
            json_data = {'conf': data}
        webserver_url = (
            'https://'
            + webserver_id
            + '.appspot.com/'
            + endpoint
        )
        # Make a POST request to IAP which then Triggers the DAG
        make_iap_request(
            webserver_url, client_id, method='POST', json=json_data)
    
    # This code is copied from
    # https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/iap/make_iap_request.py
    # START COPIED IAP CODE
    def make_iap_request(url, client_id, method='GET', **kwargs):
        """Makes a request to an application protected by Identity-Aware Proxy.
        Args:
          url: The Identity-Aware Proxy-protected URL to fetch.
          client_id: The client ID used by Identity-Aware Proxy.
          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.
        Returns:
          The page body, or raises an exception if the page couldn't be retrieved.
        """
        # Set the default timeout, if missing
        if 'timeout' not in kwargs:
            kwargs['timeout'] = 90
    
        # Obtain an OpenID Connect (OIDC) token from metadata server or using service
        # account.
        google_open_id_connect_token = id_token.fetch_id_token(Request(), client_id)
    
        # Fetch the Identity-Aware Proxy-protected URL, including an
        # Authorization header containing "Bearer " followed by a
        # Google-issued OpenID Connect token for the service account.
        resp = requests.request(
            method, url,
            headers={'Authorization': 'Bearer {}'.format(
                google_open_id_connect_token)}, **kwargs)
        if resp.status_code == 403:
            raise Exception('Service account does not have permission to '
                            'access the IAP-protected application.')
        elif resp.status_code != 200:
            raise Exception(
                'Bad response from application: {!r} / {!r} / {!r}'.format(
                    resp.status_code, resp.headers, resp.text))
        else:
            return resp.text
    # END COPIED IAP CODE
    

    Testar a função

    Para verificar se a função e o DAG funcionam conforme o esperado:

    1. Aguarde até que a função seja implantada.
    2. Faça upload de um arquivo para o bucket do Cloud Storage. Como alternativa, é possível acionar a função manualmente selecionando a ação Testar função para ela no Console do Google Cloud.
    3. Verifique a página do DAG na interface da Web do Airflow. O DAG precisa ter uma execução ativa ou já concluída.
    4. Na IU do Airflow, verifique os registros de tarefas desta execução. Você verá que a tarefa print_gcs_info gera os dados recebidos da função para os registros:
    [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
    

    A seguir