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 的函数：
- 您可以在 Cloud Run functions 中为函数配置触发器。
- 当函数被触发时，它会通过 Managed Airflow 环境的 Airflow REST API 发出请求来触发 DAG。相应请求包含事件的标识符和类型，以及事件的载荷。
- Airflow 会处理此请求并运行请求中指定的 DAG。DAG 会输出从函数传递给它的数据。
准备工作
本部分列出了准备步骤。
检查环境的网络配置
此解决方案在专用 IP 和 VPC Service Controls 配置中不起作用，因为在这些配置中，无法配置从 Cloud Run 函数到 Airflow Web 服务器的连接。
在 Managed Airflow（第 3 代）中，您可以使用另一种方法：使用 Cloud Run 函数和 Pub/Sub 消息触发 DAG。
为您的项目启用 API
控制台
启用 Managed Airflow 和 Cloud Run functions API。
启用 API 所需的角色
如需启用 API，您需要拥有 serviceusage.services.enable 权限。如果您创建了项目，则可能已经通过 Owner 角色 (roles/owner) 获得了此权限。否则，您可以通过 Service Usage Admin 角色 (roles/serviceusage.serviceUsageAdmin) 获得此权限。了解如何授予角色。
gcloud
启用 Managed Airflow 和 Cloud Run functions API：
启用 API 所需的角色
如需启用 API，您需要拥有 serviceusage.services.enable 权限。如果您创建了项目，则可能已经通过 Owner 角色 (roles/owner) 获得了此权限。否则，您可以通过 Service Usage Admin 角色 (roles/serviceusage.serviceUsageAdmin) 获得此权限。了解如何授予角色。
gcloud services enable cloudfunctions.googleapis.comcomposer.googleapis.com
允许使用 Web 服务器网络访问权限控制对 Airflow REST API 进行 API 调用
Cloud Run functions 可以通过 IPv4 或 IPv6 地址访问 Airflow REST API。
如果您不确定调用 IP 范围，请在网络服务器访问权限控制中使用默认配置选项 All IP addresses have access (default)，以免意外阻止 Cloud Run functions。您随时可以稍后配置 Web 服务器网络访问权限。
获取 Airflow Web 服务器网址
此示例向 Airflow 网络服务器端点发出 REST API 请求。您可以在 Cloud Functions 函数代码中使用 Airflow 网络服务器的网址。
控制台
在 Google Cloud 控制台中，前往环境页面。
点击您的环境的名称。
在环境详情页面上，前往环境配置标签页。
Airflow 网页界面项中列出了 Airflow 网络服务器的网址。
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 Function。
指定函数配置参数
触发器：为函数选择一个 Eventarc 触发器或多个触发器。
如需详细了解如何创建触发器，请参阅使用 Eventarc 创建触发器。 例如，您可以使用 Eventarc 触发 Cloud Storage 中的函数。
服务账号：您为触发器指定的服务账号必须具有足够的权限来触发 Managed Airflow 环境中的 DAG。
我们建议遵循最小权限原则，仅向其授予 Composer User (
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 是否按预期工作，请执行以下操作：
- 等待您的函数完成部署。
- 根据指定的触发器触发函数。您也可以手动触发函数，方法是在 Google Cloud 控制台中为该函数选择测试函数操作。
- 查看 Airflow 网页界面中的 DAG 页面。DAG 应具有一个处于活动状态或已完成的 DAG 运行。
- 在 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字段，但失败了。例如，如果您在测试函数时发送任意事件载荷，就可能会发生这种情况。 - 如果您的函数失败并显示
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 中有所不同。