הפעלת 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 בהגדרות האלה.

ב-Managed Airflow (דור 2), אפשר להשתמש בגישה אחרת: הפעלת DAG באמצעות פונקציות Cloud Run והודעות Pub/Sub.

הפעלת ממשקי 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 2, ממשק ה-API היציב ל-REST כבר מופעל כברירת מחדל. אם ה-API היציב מושבת בסביבה שלכם, צריך להפעיל את ה-API היציב ל-REST.

התרת קריאות ל-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 בקוד של Cloud Functions.

המסוף

  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 עם האזור שבו הסביבה ממוקמת.

העלאת 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_with_gcf כנקודת הכניסה.

    • (Java) כשמוסיפים קוד לדוגמה הזו, בוחרים את זמן הריצה Java 17 ומציינים את functions.TriggerDagExample כנקודת הכניסה.

הוספת דרישות

Python

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

google-auth>=2.38.0
requests>=2.34.2
functions-framework==3.*

Java

מוסיפים את יחסי התלות הבאים לקטע dependencies בקובץ pom.xml:

    <dependency>
      <groupId>com.google.apis</groupId>
      <artifactId>google-api-services-docs</artifactId>
      <version>v1-rev20250917-2.0.0</version>
    </dependency>
    <dependency>
      <groupId>com.google.api-client</groupId>
      <artifactId>google-api-client</artifactId>
      <version>2.9.0</version>
    </dependency>
    <dependency>
      <groupId>com.google.auth</groupId>
      <artifactId>google-auth-library-credentials</artifactId>
      <version>1.49.0</version>
    </dependency>
    <dependency>
      <groupId>com.google.auth</groupId>
      <artifactId>google-auth-library-oauth2-http</artifactId>
      <version>1.49.0</version>
    </dependency>

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

Python

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

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

  • אם מפעילים DAG אחר, מחליפים את הערך של המשתנה dag_id.

from __future__ import annotations

from typing import Any

from datetime import datetime, timezone
import google.auth
from google.auth.transport.requests import AuthorizedSession
import requests
import functions_framework

# Following Google Cloud 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_managed_airflow_web_server_request(
    url: str, method: str = "GET", **kwargs: Any
) -> google.auth.transport.Response:
    """
    Make a request to 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_request(web_server_url: str, airflow_version: str, dag_id: str, data: dict, logical_date: str) -> str:
    """
    Make a request to trigger a dag using the Airflow REST API.
    https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html

    Args:
      web_server_url: The URL of the Airflow web server.
      airflow_version: Major version of Airflow. Determines the API endpoint.
      dag_id: The DAG ID.
      data: Additional configuration parameters for the DAG run (json).
      logical_date: Data interval for which to run the DAG.
    """

    if airflow_version == "2":
        endpoint = f"api/v1/dags/{dag_id}/dagRuns"
    elif airflow_version == "3":
        endpoint = f"api/v2/dags/{dag_id}/dagRuns"
    else:
        raise ValueError(
          f"Invalid Airflow version: {airflow_version}. Expected: 2 or 3.")

    request_url = f"{web_server_url}/{endpoint}"
    json_data = {
        "conf": data,
        "logical_date": logical_date,
    }

    response = make_managed_airflow_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

@functions_framework.cloud_event
def trigger_dag_with_gcf(cloud_event: CloudEvent) -> None:
    """
    Entry point for the Cloud Function. Triggers a DAG and passes event data.
    """

    # cloud_event.data contains the resource payload (e.g., storage object
    # details or pub/sub body)
    event_data = {
        "id": cloud_event["id"],
        "subject": cloud_event["subject"],
        "type": cloud_event["type"],
        "data": cloud_event.data
    }

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

    # TODO(developer): If your environment uses Airflow 3, replace with "3"
    airflow_major_version = "2"

    # Replace with the ID of the DAG that you want to run.
    dag_id = "composer_sample_trigger_response_dag"

    # The data interval for which to run the DAG
    # Format example: "2026-07-15T15:00:00Z"
    now = datetime.now(timezone.utc)
    logical_date = now.strftime("%Y-%m-%dT%H:%M:%SZ")

    trigger_dag_request(web_server_url, airflow_major_version, dag_id, event_data, logical_date)

Java

מכניסים את הקוד הבא לקובץ TriggerDagExample.java (צריך להכניס את הקובץ הזה לתיקייה src/main/java/gcfv2/):

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

  • אם מפעילים DAG אחר, מחליפים את הערך של המשתנה dagName.

package gcfv2;

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.CloudEventsFunction;
import com.google.gson.Gson;
import io.cloudevents.CloudEvent;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.logging.Logger;
import java.util.HashMap;
import java.util.Map;

/**
 * Function that triggers an Airflow DAG in response to an event ad passes data.
 */
public class TriggerDagExample implements CloudEventsFunction {
  private static final Logger logger = Logger.getLogger(TriggerDagExample.class.getName());

  @Override
  public void accept(CloudEvent event) throws Exception{

    // 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)"
    String webServerUrl = "https://example-airflow-ui-url-dot-us-central1.composer.googleusercontent.com";
    // TODO(developer): If your environment uses Airflow 3, replace with "3"
    String majorAirflowVersion = "2";

    String apiVersion = switch (majorAirflowVersion) {
      case "2" -> "v1";
      case "3" -> "v2";
      default  -> throw new IllegalArgumentException("Invalid Airflow version: " + majorAirflowVersion);
    };

    String dagName = "composer_sample_trigger_response_dag";
    String url = String.format("%s/api/%s/dags/%s/dagRuns", webServerUrl, apiVersion, dagName);

    logger.info(String.format("Triggering DAG %s as a result of an event on the object %s.",
      dagName, event.getSubject()));
    logger.info(String.format("Triggering DAG through 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, Object> conf = new HashMap<>();

    conf.put("id", event.getId());
    conf.put("subject", event.getSubject());
    conf.put("type", event.getType());

    if (event.getData() != null) {
      String dataJson = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
      Gson gson = new Gson();
      Map<String, Object> dataMap = gson.fromJson(dataJson, Map.class);
      conf.put("data", dataMap);
    }

    String currentUtcTime = Instant.now().toString();

    Map<String, Object> json = new HashMap<>();
    json.put("conf", conf);
    json.put("logical_date", currentUtcTime);

    HttpContent content = new JsonHttpContent(new GsonFactory(), json);
    HttpRequest request = requestFactory.buildPostRequest(new GenericUrl(url), content);
    request.getHeaders().setContentType("application/json");

    HttpResponse response = null;
    try {
      response = request.execute();
      int statusCode = response.getStatusCode();
      logger.info("Response code: " + statusCode);
      logger.info(response.parseAsString());
    } catch (HttpResponseException e) {
      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());
    } finally {
      // Safely close and release the HTTP connection pool resource
      if (response != null) {
        try {
          response.disconnect();
        } catch (Exception e) {
          logger.warning("Failed to disconnect response: " + e.getMessage());
        }
      }
    }
  }
}

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

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

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

פקודה לדוגמה לבדיקת הפונקציה:

curl -X POST "https://service-id.region.run.app" \
-H "Authorization: bearer $(gcloud auth print-identity-token)" \
  -X POST \
  -H "Content-Type: application/json" \
  -H "ce-id: 1234567890" \
  -H "ce-specversion: 1.0" \
  -H "ce-type: google.cloud.storage.object.v1.finalized" \
  -H "ce-source: //storage.googleapis.com/projects/_/buckets/example-bucket" \
  -d '{
    "name": "example-file.csv",
    "bucket": "example-bucket"
  }'

פלט לדוגמה:

[2026-07-14, 15:10:12 UTC] {subprocess.py:88} INFO - Running command: ['/usr/bin/bash', '-c', "echo {'data': {'name': 'example-file.csv', 'bucket': 'example-bucket'}, 'id': '1234567890', 'type': 'google.cloud.storage.object.v1.finalized'}"]
[2026-07-14, 15:10:12 UTC] {subprocess.py:99} INFO - Output:
[2026-07-14, 15:10:12 UTC] {subprocess.py:106} INFO - {data: {name: example-file.csv, bucket: my-bucket}, id: 1234567890, type: google.cloud.storage.object.v1.finalized}
[2026-07-14, 15:10:12 UTC] {subprocess.py:110} INFO - Command exited with return code 0

[2026-07-15, 10:06:32 UTC] {subprocess.py:88} INFO - Running command: ['/usr/bin/bash', '-c', "echo {'id': '1234567890', 'subject': 'objects/example-file.csv', 'type': 'google.cloud.storage.object.v1.finalized', 'data': {'name': 'example-file.csv', 'bucket': 'example-bucket'}}"]
[2026-07-15, 10:06:32 UTC] {subprocess.py:99} INFO - Output:
[2026-07-15, 10:06:32 UTC] {subprocess.py:106} INFO - {id: 1234567890, subject: objects/example-file.csv, type: google.cloud.storage.object.v1.finalized, data: {name: example-file.csv, bucket: example-bucket}}
[2026-07-15, 10:06:32 UTC] {subprocess.py:110} INFO - Command exited with return code 0

פתרון בעיות:

  • אם הפונקציה נכשלת עם שגיאה NullPointerException: Null data ודוח הקריסות מצביע על הפונקציה BackgroundFunctionExecutor.parseLegacyEvent, זה אומר שהאירוע שהתקבל על ידי הפונקציה לא כולל כותרות מטא-נתונים סטנדרטיות CloudEvent. הפונקציה מניחה שאתם שולחים אירוע ברקע מדור קודם, מנסה לנתח את השדה data ממנו ונכשלת. לדוגמה, זה יכול לקרות אם שולחים מטען ייעודי (payload) של אירוע שרירותי כשבודקים את הפונקציה.
  • אם הפונקציה נכשלת עם השגיאה 500 Internal Server Error: The server encountered an internal error and was unable to complete your request., צריך לבדוק שוב את הערך של המשתנה airflow_major_version. המשתנה הזה קובע את נקודת הקצה ל-API בארכיטקטורת REST של Airflow, שהיא שונה ב-Airflow 2 וב-Airflow 3.

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