使用 Cloud Run 函式和 Airflow REST API 觸發 Managed Service for Apache Airflow DAG

Managed Airflow (第 3 代) | Managed Airflow (第 2 代) | Managed Airflow (舊版第 1 代)

本頁說明如何使用 Cloud Run 函式，根據事件觸發 Managed Service for Apache Airflow DAG。

Apache Airflow 的設計是定期執行 DAG，但您也可以在發生事件時觸發 DAG。其中一種做法是使用 Cloud Run 函式，在發生指定事件時觸發 Managed Airflow DAG。

您也能執行下列操作：

本指南中的範例會示範函式如何觸發 DAG 來回應事件：

  1. 您可以在 Cloud Run 函式中設定函式的觸發條件。
  2. 函式觸發後，會透過 Managed Airflow 環境的 Airflow REST API 提出要求，觸發 DAG。要求包含事件的 ID 和類型，以及事件的酬載。
  3. Airflow 會處理這項要求，並執行要求中指定的 DAG。 DAG 會輸出從函式傳遞給 DAG 的資料。

事前準備

本節列出準備步驟。

檢查環境的網路設定

這個解決方案不適用於私人 IP 和 VPC Service Controls 設定，因為在這些設定中，無法設定從 Cloud Run 函式到 Airflow 網路伺服器的連線。

在 Managed Airflow (第 3 代) 中，您可以使用其他方法：使用 Cloud Run 函式和 Pub/Sub 訊息觸發 DAG。

為專案啟用 API

控制台

啟用 Managed Airflow 和 Cloud Run functions API。

啟用 API 時所需的角色

如要啟用 API，您必須具備 serviceusage.services.enable 權限。如果您建立了專案，可能已透過「擁有者」角色 (roles/owner) 取得這項權限。否則，您可以透過「服務使用情形管理員」角色 (roles/serviceusage.serviceUsageAdmin) 取得這項權限。瞭解如何授予角色。

啟用 API

gcloud

啟用 Managed Airflow 和 Cloud Run functions API：

啟用 API 時所需的角色

如要啟用 API，您必須具備 serviceusage.services.enable 權限。如果您建立了專案，可能已透過「擁有者」角色 (roles/owner) 取得這項權限。否則，您可以透過「服務使用情形管理員」角色 (roles/serviceusage.serviceUsageAdmin) 取得這項權限。瞭解如何授予角色。 

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

使用網路伺服器網路存取控管，允許對 Airflow REST API 進行 API 呼叫

Cloud Run 函式可透過 IPv4 或 IPv6 位址連線至 Airflow REST API。

如果不確定呼叫 IP 範圍，請使用「Webserver Access Control」中的預設設定選項 All IP addresses have access (default)，以免不慎封鎖 Cloud Run functions。您隨時可以稍後設定網路伺服器網路存取權。

取得 Airflow 網路伺服器網址

這個範例會向 Airflow 網路伺服器端點發出 REST API 要求。您會在 Cloud 函式程式碼中使用 Airflow 網路伺服器的網址。

控制台

  1. 前往 Google Cloud 控制台的「Environments」(環境) 頁面。

    前往「環境」

  2. 按一下環境名稱。

  3. 在「環境詳細資料」頁面中，前往「環境設定」分頁。

  4. Airflow 網路伺服器的網址會列在「Airflow web UI」(Airflow 網頁版 UI) 項目中。

gcloud

執行下列指令：

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

更改項目：

  • 將 ENVIRONMENT_NAME 替換為環境的名稱。
  • 將 LOCATION 替換為環境所在的區域。

將 DAG 上傳至環境

將 DAG 上傳至環境。 下列 DAG 範例會輸出收到的 DAG 執行設定。您會從函式觸發這個 DAG，該函式會在稍後建立。

Airflow 3 

import datetime

import airflow
from airflow.providers.standard.operators.bash 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 }}}}')

Airflow 2 

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 functions 或 Cloud Run 支援的偏好語言部署函式。本教學課程將示範以 Python 和 Java 實作的 Cloud 函式。

指定函式設定參數

  • 觸發條件：為函式選取一個或多個 Eventarc 觸發條件。

    如要進一步瞭解如何建立觸發條件，請參閱使用 Eventarc 建立觸發條件。 舉例來說，您可以使用 Eventarc，透過 Cloud Storage 觸發函式。

  • 服務帳戶：您為觸發條件指定的服務帳戶必須具備足夠的權限，才能在 Managed Airflow 環境中觸發 DAG。

    建議您遵循最低權限原則，只授予該帳戶「Composer 使用者」 (composer.user) 角色。如要進一步瞭解如何設定權限，請參閱「Cloud Run 目標的角色和權限」。

  • 函式進入點：

    • (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

在 pom.xml 的 dependencies 部分新增下列依附元件：

    <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 網頁伺服器位址。

  • (Airflow 3) 將 airflow_major_version 變數的值替換為 3，這是環境中的 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 網頁伺服器位址。

  • (Airflow 3) 將 majorAirflowVersion 變數的值替換為 3，這是環境中的 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. 根據指定的觸發條件觸發函式。您也可以在 Google Cloud 控制台中選取函式的「測試函式」動作，手動觸發函式。
  3. 在 Airflow 網頁介面中查看 DAG 頁面。DAG 應有一個有效或已完成的 DAG 執行作業。
  4. 在 Airflow UI 中，查看這項執行的工作記錄。您應該會看到 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 欄位，但會失敗。舉例來說，測試函式時，如果傳送任意事件酬載，就可能發生這種情況。
  • 如果函式失敗並顯示 500 Internal Server Error: The server encountered an internal error and was unable to complete your request.，請仔細檢查 airflow_major_version 變數的值。這個變數會決定 Airflow REST API 端點，Airflow 2 和 Airflow 3 的端點不同。

後續步驟