Memicu DAG Cloud Composer dengan Cloud Functions dan Airflow REST API

Cloud Composer 1 | Cloud Composer 2

Halaman ini menjelaskan cara menggunakan Cloud Functions untuk memicu DAG Cloud Composer sebagai respons terhadap peristiwa.

Catatan: Anda juga dapat memicu DAG menggunakan Airflow REST API saja.

Apache Airflow dirancang untuk menjalankan DAG sesuai jadwal rutin, tetapi Anda juga dapat memicu DAG sebagai respons terhadap peristiwa. Salah satu cara untuk melakukannya adalah dengan menggunakan Cloud Functions untuk memicu DAG Cloud Composer saat peristiwa tertentu terjadi.

Contoh dalam panduan ini menjalankan DAG setiap kali terjadi perubahan di bucket Cloud Storage. Perubahan pada objek dalam bucket akan memicu fungsi. Fungsi ini membuat permintaan ke Airflow REST API di lingkungan Cloud Composer Anda. Airflow memproses permintaan ini dan menjalankan DAG. DAG menghasilkan informasi tentang perubahan tersebut.

Catatan: Panduan alternatif menunjukkan cara membuat fungsi yang memicu DAG saat pesan dikirim ke topik Pub/Sub.

Sebelum memulai

Memeriksa konfigurasi jaringan lingkungan Anda

Solusi ini tidak berfungsi dalam konfigurasi IP Pribadi dan Kontrol Layanan VPC karena konfigurasi ini tidak dapat dikonfigurasi dari Cloud Functions ke server web Airflow.

Di Cloud Composer 2, Anda dapat menggunakan pendekatan lain: Memicu DAG menggunakan Cloud Functions dan Pesan Pub/Sub

Mengaktifkan API untuk project Anda

Konsol

Enable the Cloud Composer and Cloud Functions APIs.

Enable the APIs

gcloud

Enable the Cloud Composer and Cloud Functions APIs:

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

Mengaktifkan Airflow REST API

Bergantung pada versi Airflow Anda:

• Untuk Airflow 2, REST API stabil sudah diaktifkan secara default. Jika lingkungan Anda menonaktifkan API stabil, aktifkan REST API stabil.
• Untuk Airflow 1, aktifkan REST API eksperimental.

Izinkan panggilan API ke Airflow REST API menggunakan Kontrol Akses Webserver

Cloud Functions dapat menjangkau Airflow REST API menggunakan alamat IPv4 atau IPv6.

Jika Anda tidak yakin apa yang akan menjadi rentang IP panggilan, gunakan opsi konfigurasi default di Webserver Access Control All IP addresses have access (default) untuk tidak memblokir Cloud Functions Anda secara tidak sengaja.

Membuat bucket Cloud Storage

Contoh ini memicu DAG sebagai respons terhadap perubahan di bucket Cloud Storage. buat bucket baru untuk digunakan dalam contoh ini.

Catatan: Jangan gunakan bucket lingkungan Cloud Composer karena dapat menyebabkan fungsi dipicu beberapa kali per detik. Untuk tujuan contoh ini, sebaiknya buat bucket terpisah.

Mendapatkan URL server web Airflow

Contoh ini membuat permintaan REST API ke endpoint server web Airflow. Gunakan bagian URL antarmuka web Airflow sebelum .appspot.com di kode Cloud Function Anda.

Konsol

1. Di konsol Google Cloud, buka halaman Environments.

   Buka Lingkungan

2. Klik nama lingkungan Anda.

3. Di halaman Environment details, buka tab Environment configuration.

4. URL server web Airflow tercantum dalam item UI web Airflow.

gcloud

Jalankan perintah berikut:

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

Ganti:

• ENVIRONMENT_NAME dengan nama lingkungan.
• LOCATION dengan region tempat lingkungan berada.

Mendapatkan client_id proxy IAM

Perhatian: Panduan ini ditujukan untuk Cloud Composer 1 dan langkah-langkah yang dijelaskan tidak akan berfungsi di Cloud Composer 2. Sebagai gantinya, ikuti petunjuk di halaman Memicu DAG dengan Cloud Functions untuk Cloud Composer 2.

Untuk membuat permintaan ke endpoint Airflow REST API, fungsi ini memerlukan client ID dari proxy Identity and Access Management yang melindungi server web Airflow.

Cloud Composer tidak memberikan informasi ini secara langsung. Sebagai gantinya, buat permintaan yang tidak diautentikasi ke server web Airflow dan ambil client ID dari URL alihan:

cURL

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

Ganti AIRFLOW_URL dengan URL antarmuka web Airflow.

Pada output, telusuri string setelah client_id. Contoh:

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

Python

Simpan kode berikut di file bernama get_client_id.py. Isi nilai untuk project_id, location, dan composer_environment, lalu jalankan kode di Cloud Shell atau lingkungan lokal Anda.

# 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])

Mengupload DAG ke lingkungan Anda

Upload DAG ke lingkungan Anda. Contoh DAG berikut menghasilkan output konfigurasi run DAG yang diterima. Anda memicu DAG ini dari fungsi yang akan Anda buat nanti dalam panduan ini.

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 }}"
    )

Men-deploy Cloud Function yang memicu DAG

Anda dapat men-deploy Cloud Function menggunakan bahasa pilihan Anda yang didukung oleh Cloud Functions atau Cloud Run. Tutorial ini menunjukkan Cloud Function yang diimplementasikan dalam Python dan Java.

Menentukan parameter konfigurasi Cloud Function

• Pemicu. Untuk contoh ini, pilih pemicu yang berfungsi saat objek baru dibuat di dalam bucket, atau objek yang sudah ada akan ditimpa.

  • Jenis Pemicu. yang sesuai di Cloud Storage.

  • Jenis Peristiwa. Selesaikan / Buat.

  • Bucket. Pilih bucket yang harus memicu fungsi ini.

  • Coba lagi jika gagal. Sebaiknya nonaktifkan opsi ini untuk tujuan contoh ini. Jika Anda menggunakan fungsi Anda sendiri di lingkungan produksi, aktifkan opsi ini untuk menangani error sementara.

• Akun layanan runtime, di bagian Setelan runtime, build, koneksi dan keamanan. Gunakan salah satu opsi berikut, bergantung pada preferensi Anda:

  • Pilih Akun layanan default Compute Engine. Dengan izin IAM default, akun ini dapat menjalankan fungsi yang mengakses lingkungan Cloud Composer.

  • Buat akun layanan kustom yang memiliki peran Composer User dan tentukan sebagai akun layanan runtime untuk fungsi ini. Opsi ini mengikuti prinsip hak istimewa minimum.

• Runtime dan titik entri, pada langkah Code. Saat menambahkan kode untuk contoh ini, pilih runtime Python 3.7 atau yang lebih baru dan tentukan trigger_dag sebagai titik entri.

Tambahkan persyaratan

Tentukan dependensi dalam file requirements.txt:

requests-toolbelt==1.0.0
google-auth==2.19.1
google-cloud-pubsub==2.17.0

Masukkan kode berikut ke file main.py dan buat penggantian berikut:

• Ganti nilai variabel client_id dengan nilai client_id yang Anda peroleh sebelumnya.

• Ganti nilai variabel webserver_id dengan ID project tenant Anda, yang merupakan bagian dari URL antarmuka web Airflow sebelum .appspot.com. Anda sudah mendapatkan URL antarmuka web Airflow sebelumnya.

• Tentukan versi Airflow REST API yang Anda gunakan:

  • Jika Anda menggunakan Airflow REST API yang stabil, tetapkan variabel USE_EXPERIMENTAL_API ke False.
  • Jika menggunakan Airflow REST API eksperimental, Anda tidak perlu melakukan perubahan apa pun. Variabel USE_EXPERIMENTAL_API sudah ditetapkan ke True.

Catatan: Akun layanan terkadang memiliki alamat email yang panjangnya lebih dari 64 karakter. Untuk mengakses Airflow REST API dengan akun layanan tersebut, gunakan solusi yang dijelaskan dalam Mengakses Airflow REST API menggunakan akun layanan.

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

Menguji fungsi

Untuk memeriksa apakah fungsi dan DAG berfungsi sebagaimana mestinya:

1. Tunggu hingga fungsi Anda di-deploy.
2. Mengupload file ke bucket Cloud Storage Anda. Sebagai alternatif, Anda dapat memicu fungsi secara manual dengan memilih tindakan Test the function untuknya di Konsol Google Cloud.
3. Periksa halaman DAG di antarmuka web Airflow. DAG harus memiliki satu pengoperasian DAG yang aktif atau sudah selesai.
4. Di UI Airflow, periksa log tugas untuk operasi ini. Anda akan melihat bahwa tugas print_gcs_info menghasilkan output data yang diterima dari fungsi ke log:

[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

Langkah selanjutnya

• Mengakses UI Airflow
• Mengakses Airflow REST API
• Menulis DAG
• Menulis Cloud Functions
• Pemicu Cloud Storage