Managed Airflow (3세대) | Managed Airflow (2세대) | Managed Airflow (기존 1세대)
이 페이지에서는 Cloud Run Functions를 사용하여 이벤트에 대한 응답으로 Managed Service for Apache Airflow DAG를 트리거하는 방법을 설명합니다.
Apache Airflow는 DAG를 정기적으로 실행하도록 설계되어 있지만 이벤트에 대한 응답으로 DAG를 트리거할 수도 있습니다. 이를 수행하는 한 가지 방법은 Cloud Run Functions를 사용하여 지정된 이벤트가 발생할 때 관리형 Airflow DAG를 트리거하는 것입니다.
다음과 같은 작업도 할 수 있습니다.
- Airflow REST API만 사용하여 DAG를 트리거합니다.
- 메시지가 Pub/Sub 주제에 푸시될 때 DAG를 트리거하는 함수를 만듭니다.
이 가이드의 예시에서는 이벤트에 대한 응답으로 DAG를 트리거하는 함수를 보여줍니다.
- Cloud Run Functions에서 함수의 트리거를 구성합니다.
- 함수가 트리거되면 Managed Airflow 환경의 Airflow REST API를 통해 DAG를 트리거하는 요청을 보냅니다. 요청에는 이벤트의 식별자 및 유형과 이벤트의 페이로드가 포함됩니다.
- Airflow가 이 요청을 처리하고 요청에 지정된 DAG를 실행합니다. DAG는 함수에서 전달된 데이터를 출력합니다.
시작하기 전에
이 섹션에는 준비 단계가 나와 있습니다.
환경의 네트워킹 구성 확인
비공개 IP 및 VPC 서비스 제어 구성에서는 Cloud Run Functions에서 Airflow 웹 서버로의 연결을 구성할 수 없으므로 해당 구성에서는 이 솔루션이 작동하지 않습니다.
관리형 Airflow (2세대)에서 또 다른 접근 방법인 Cloud Run Functions 및 Pub/Sub 메시지를 사용하여 DAG 트리거를 사용할 수 있습니다.
프로젝트에 API 사용 설정
콘솔
Managed Airflow 및 Cloud Run Functions API를 사용 설정합니다.
API 사용 설정에 필요한 역할
API를 사용 설정하려면 serviceusage.services.enable 권한이 필요합니다. 프로젝트를 만든 경우 소유자 역할 (roles/owner)을 통해 이 권한을 이미 보유하고 있을 가능성이 높습니다. 그렇지 않은 경우 서비스 사용량 관리자 역할 (roles/serviceusage.serviceUsageAdmin)을 통해 이 권한을 얻을 수 있습니다. 역할을 부여하는 방법을 알아보세요.
gcloud
관리형 Airflow 및 Cloud Run Functions API를 사용 설정합니다.
API 사용 설정에 필요한 역할
API를 사용 설정하려면 serviceusage.services.enable 권한이 필요합니다. 프로젝트를 만든 경우 소유자 역할 (roles/owner)을 통해 이 권한을 이미 보유하고 있을 가능성이 높습니다. 그렇지 않은 경우 서비스 사용량 관리자 역할 (roles/serviceusage.serviceUsageAdmin)을 통해 이 권한을 얻을 수 있습니다. 역할을 부여하는 방법을 알아보세요.
gcloud services enable cloudfunctions.googleapis.comcomposer.googleapis.com
Airflow REST API 사용 설정
Airflow 2의 경우 안정적인 REST API가 기본으로 사용 설정되어 있습니다. 환경에서 안정적인 API를 사용하지 않는 경우 안정적인 REST API를 사용 설정합니다.
웹 서버 네트워크 액세스 제어를 사용하여 Airflow REST API에 대한 API 호출 허용
Cloud Run Functions는 IPv4 또는 IPv6 주소 를 통해 Airflow REST API에 연결할 수 있습니다.
호출 IP 범위를 모르는 경우 웹 서버 액세스 제어 의 기본 구성 옵션인 All IP addresses have access (default)를 사용하여 Cloud Run Functions를 실수로 차단하지 않도록 합니다. 나중에 언제든지
웹 서버 네트워크 액세스를 구성할 수 있습니다.
Airflow 웹 서버 URL 가져오기
이 예시에서는 Airflow 웹 서버 엔드포인트에 REST API 요청을 보냅니다. Cloud 함수 코드에서 Airflow 웹 서버 URL을 사용합니다.
콘솔
콘솔 Google Cloud 에서 환경 페이지로 이동합니다.
환경 이름을 클릭합니다.
환경 세부정보 페이지에서 환경 구성 탭으로 이동합니다.
Airflow 웹 서버의 URL이 Airflow 웹 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 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 웹 서버 주소로 바꿉니다.다른 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가 의도한 대로 작동하는지 확인하기 위해 다음 안내를 따르세요.
- 함수가 배포될 때까지 기다립니다.
- 지정된 트리거에 따라 함수를 트리거합니다. 콘솔 Google Cloud 에서 함수 테스트 작업을 선택하여 함수를 수동으로 트리거할 수도 있습니다.
- Airflow 웹 인터페이스에서 DAG 페이지를 확인합니다. DAG는 DAG 실행이 활성 상태 또는 이미 완료된 상태여야 합니다.
- 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에서 서로 다릅니다.