הפעלת DAGs של Managed Service for Apache Airflow באמצעות פונקציות Cloud Run ו-API בארכיטקטורת REST של Airflow

‫Managed Airflow (דור 3) | Managed Airflow (דור 2) | Managed Airflow (דור 1 מדור קודם)

בדף הזה מוסבר איך להשתמש בפונקציות של Cloud Run כדי להפעיל DAG של Managed Service for Apache Airflow בתגובה לאירועים.

‫Apache Airflow מיועד להרצת DAGs בלוח זמנים קבוע, אבל אפשר גם להפעיל DAGs בתגובה לאירועים. אחת הדרכים לעשות את זה היא להשתמש בפונקציות Cloud Run כדי להפעיל DAGs מנוהלים של Airflow כשמתרחש אירוע ספציפי.

אפשר גם:

בדוגמה שבמדריך הזה מוצגת פונקציה שמפעילה DAG בתגובה לאירוע:

  1. מגדירים טריגרים לפונקציה ב-Cloud Run Functions.
  2. כשהפונקציה מופעלת, היא שולחת בקשה להפעלת DAG דרך API בארכיטקטורת REST של Airflow בסביבת Managed Airflow. הבקשה מכילה את המזהה והסוג של האירוע, ואת מטען הייעודי (payload) של האירוע.
  3. מערכת Airflow מעבדת את הבקשה ומריצה את ה-DAG שצוין בבקשה. ה-DAG מוציא את הנתונים שהועברו אליו מהפונקציה.

לפני שמתחילים

בקטע הזה מפורטים שלבי ההכנה.

בדיקת הגדרות הרשת בסביבה

הפתרון הזה לא פועל בהגדרות של כתובות IP פרטיות ו-VPC Service Controls, כי אי אפשר להגדיר קישוריות מפונקציות של Cloud Run לשרת האינטרנט של Airflow בהגדרות האלה.

הפעלת ממשקי API בפרויקט

המסוף

מפעילים את ממשקי ה-API של Managed Airflow ושל פונקציות Cloud Run.

תפקידים שנדרשים להפעלת ממשקי API

כדי להפעיל ממשקי API, נדרשת ההרשאה serviceusage.services.enable. אם יצרתם את הפרויקט, סביר להניח שכבר יש לכם את ההרשאה הזו דרך התפקיד 'בעלים' (roles/owner). אחרת, תוכלו לקבל את ההרשאה הזו דרך התפקיד 'אדמין בממשק Service Usage' (roles/serviceusage.serviceUsageAdmin). איך מקצים תפקידים

הפעלת ממשקי ה-API

gcloud

מפעילים את ממשקי ה-API של פונקציות Managed Airflow ו-Cloud Run:

תפקידים שנדרשים להפעלת ממשקי API

כדי להפעיל ממשקי API, נדרשת ההרשאה serviceusage.services.enable. אם יצרתם את הפרויקט, סביר להניח שכבר יש לכם את ההרשאה הזו דרך התפקיד 'בעלים' (roles/owner). אחרת, תוכלו לקבל את ההרשאה הזו דרך התפקיד 'אדמין בממשק Service Usage' (roles/serviceusage.serviceUsageAdmin). איך מקצים תפקידים 

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

הפעלת Airflow API בארכיטקטורת REST

בהתאם לגרסת Airflow:

התרת קריאות ל-API בארכיטקטורת REST של Airflow באמצעות בקרת גישה לרשת של שרת האינטרנט

פונקציות Cloud Run יכולות לפנות אל Airflow API בארכיטקטורת REST באמצעות כתובת IPv4 או IPv6.

אם אתם לא בטוחים מה יהיה טווח כתובות ה-IP של הקריאה, כדאי להשתמש באפשרות ברירת המחדל להגדרה בWebserver Access Control, שהיא All IP addresses have access (default), כדי שלא תחסמו בטעות את הפונקציות של Cloud Run. תמיד אפשר להגדיר גישה לרשת של שרת האינטרנט מאוחר יותר.

איך מוצאים את כתובת ה-URL של שרת האינטרנט של Airflow

בדוגמה הזו מתבצעות בקשות ל-API בארכיטקטורת REST לנקודת הקצה של שרת האינטרנט של Airflow. משתמשים בחלק של כתובת ה-URL של ממשק האינטרנט של Airflow לפני .appspot.com בקוד של Cloud Function.

המסוף

  1. נכנסים לדף Environments במסוף Google Cloud .

    מעבר אל Environments

  2. לוחצים על שם הסביבה.

  3. בדף Environment details, עוברים לכרטיסייה Environment configuration.

  4. כתובת ה-URL של שרת האינטרנט של Airflow מופיעה בפריט Airflow web UI.

gcloud

מריצים את הפקודה הבאה:

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

מחליפים את:

  • ‫ENVIRONMENT_NAME בשם הסביבה.
  • ‫LOCATION עם האזור שבו הסביבה ממוקמת.

קבלת מזהה הלקוח של שרת ה-IAM proxy

כדי לשלוח בקשה לנקודת קצה ל-API בארכיטקטורת REST של Airflow, הפונקציה צריכה את מזהה הלקוח של שרת proxy לניהול זהויות והרשאות גישה שמגן על שרת האינטרנט של Airflow.

ב-Managed Airflow, המידע הזה לא זמין באופן ישיר. במקום זאת, שולחים בקשה לא מאומתת לשרת האינטרנט של Airflow ומתעדים את מזהה הלקוח מכתובת ה-URL להפניה אוטומטית:

cURL 

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

מחליפים את AIRFLOW_URL בכתובת ה-URL של ממשק האינטרנט של Airflow.

בפלט, מחפשים את המחרוזת שאחרי client_id. לדוגמה:

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

Python

שומרים את הקוד הבא בקובץ בשם get_client_id.py. ממלאים את הערכים של project_id, location ו-composer_environment, ואז מריצים את הקוד ב-Cloud Shell או בסביבה המקומית. 

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

העלאת DAG לסביבה

העלאת DAG לסביבה הפלט של ה-DAG הבא הוא הגדרת ההרצה של ה-DAG שהתקבלה. אפשר להפעיל את ה-DAG הזה מפונקציה שיוצרים בהמשך המדריך הזה.

import datetime

import airflow
from airflow.operators.bash_operator import BashOperator

with airflow.DAG(
        'composer_sample_trigger_response_dag',
        start_date=datetime.datetime(2026, 1, 1),
        # Not scheduled, trigger only
        schedule=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 }}}}')

פריסת פונקציה שמפעילה את ה-DAG

אפשר לפרוס פונקציה באמצעות השפה המועדפת שנתמכת על ידי פונקציות Cloud Run או Cloud Run. במדריך הזה נסביר איך להטמיע Cloud Function ב-Python וב-Java.

ציון פרמטרים להגדרת פונקציה

  • טריגר: בוחרים טריגר אחד או יותר של Eventarc לפונקציה.

    מידע נוסף על יצירת טריגרים זמין במאמר יצירת טריגרים באמצעות Eventarc. לדוגמה, אפשר להפעיל פונקציות מ-Cloud Storage באמצעות Eventarc.

  • חשבון שירות: לחשבון השירות שאתם מציינים לטריגר צריכות להיות הרשאות מספיקות להפעלת DAG בסביבות Managed Airflow.

    מומלץ לפעול לפי העיקרון של הרשאות מינימליות ולהעניק לו רק את התפקיד Composer User (composer.user). מידע נוסף על הגדרת הרשאות זמין במאמר תפקידים והרשאות ליעדים ב-Cloud Run.

  • Function entry point:

    • ‫(Python) כשמוסיפים קוד לדוגמה הזו, בוחרים את זמן הריצה Python 3.10 או גרסה מאוחרת יותר ומציינים את trigger_dag כנקודת הכניסה.

הוספת דרישות

מציינים את יחסי התלות בקובץ requirements.txt:

requests-toolbelt==1.0.0
google-auth==2.38.0
google-cloud-pubsub==2.28.0

הוספת קוד פונקציה

מוסיפים את הקוד הבא לקובץ main.py ומבצעים את ההחלפות הבאות:

  • מחליפים את הערך של המשתנה client_id בערך client_id שקיבלתם קודם.

  • מחליפים את הערך של המשתנה webserver_id במזהה פרויקט הדייר, שהוא חלק מכתובת ה-URL של ממשק האינטרנט של Airflow לפני .appspot.com. כתובת ה-URL של ממשק האינטרנט של Airflow הופיעה קודם.

  • מציינים את הגרסה של Airflow API בארכיטקטורת REST שבה משתמשים:

    • אם אתם משתמשים ב-Airflow API בארכיטקטורת REST היציב, צריך להגדיר את המשתנה USE_EXPERIMENTAL_API לערך False.
    • אם אתם משתמשים ב-Airflow API בארכיטקטורת REST הניסיוני, לא צריך לבצע שינויים. המשתנה USE_EXPERIMENTAL_API כבר מוגדר לערך 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

בדיקת הפונקציה

כדי לוודא שהפונקציה ו-DAG פועלים כמצופה:

  1. מחכים עד שהפונקציה תופעל.
  2. מפעילים את הפונקציה בהתאם לטריגר שצוין. אפשר גם להפעיל את הפונקציה באופן ידני על ידי בחירה בפעולה Test the function (בדיקת הפונקציה) עבורה ב- Google Cloud console.
  3. בודקים את דף ה-DAG בממשק האינטרנט של Airflow. ל-DAG צריך להיות הפעלה אחת פעילה או שכבר הסתיימה.
  4. בממשק המשתמש של Airflow, בודקים את יומני המשימות של ההרצה הזו. אפשר לראות שהנתונים שהתקבלו מהפונקציה מופיעים ביומנים של המשימה print_gcs_info:

פלט לדוגמה:

[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

המאמרים הבאים