使用 Cloud Functions 和 Pub/Sub 消息触发 DAG

Managed Airflow（第 3 代） | Managed Airflow（第 2 代） | Managed Airflow（旧版第 1 代）

注意： 此页面尚未针对 Managed Airflow（第 3 代）进行修订，显示的是 Managed Airflow（第 2 代）的内容。

本页面将指导您创建一个基于事件的推送架构，方法是触发 Managed Airflow DAG 以响应 Pub/Sub 主题更改。本教程中的示例演示了如何处理 Pub/Sub 管理的完整周期（包括订阅管理），作为 DAG 流程的一部分。它适用于某些常见的使用场景，即您需要触发 DAG，但不想设置额外的访问权限。

例如，如果您出于安全考虑而不希望提供对 Managed Airflow 环境的直接访问权限，则可以使用通过 Pub/Sub 发送的消息作为解决方案。您可以配置一个 Cloud Run 函数，该函数会创建 Pub/Sub 消息并将其发布到 Pub/Sub 主题。然后，您可以创建一个 DAG，该 DAG 会拉取 Pub/Sub 消息，然后处理这些消息。

注意： 作为替代方法，您可以仅使用 Airflow REST API触发 DAG，或者您可能希望在特定事件发生时使用 Cloud Run 函数触发 DAG。

在此特定示例中，您将创建一个 Cloud Run 函数并部署两个 DAG。第一个 DAG 会拉取 Pub/Sub 消息，并根据 Pub/Sub 消息内容触发第二个 DAG。

本教程假定您熟悉 Python 和 Google Cloud 控制台。

目标

目标：

• 创建相关资源并准备环境。
• 上传订阅 Pub/Sub 主题的 DAG。
• 配置并部署一个 Cloud Run 函数，该函数会将 Pub/Sub 消息推送到 Pub/Sub 主题。
• 测试函数。

费用

本教程使用的以下收费组件： Google Cloud

• Managed Airflow（另请参阅 额外费用）
• Pub/Sub
• Cloud Run functions

完成本教程后，您可以通过删除您创建的资源来避免继续计费。如需了解详情，请参阅清理。

准备工作

在本教程中，您需要一个 Google Cloud 项目。 请按以下方式配置项目：

1. 在 Google Cloud 控制台中，选择或创建一个项目：

   前往“项目选择器”

2. 确保您的项目已启用结算功能。了解如何检查项目是否已启用结算功能。

3. 确保您的 Google Cloud 项目用户拥有以下角色，以便创建必要的资源：

   • Service Account User (roles/iam.serviceAccountUser)
   • Pub/Sub Editor (roles/pubsub.editor)
   • Environment and Storage Object Administrator (roles/composer.environmentAndStorageObjectAdmin)
   • Cloud Run functions Admin (roles/cloudfunctions.admin)
   • Logs Viewer (roles/logging.viewer)

4. 确保运行 Cloud Run 函数的 服务账号 在您的项目中拥有足够的权限来访问 Pub/Sub。默认情况下，Cloud Run functions 使用 App Engine 默认服务账号。 此服务帐号具有 Editor 角色，该角色在本教程中拥有足够的权限。

为您的项目启用 API

控制台

启用 Managed Airflow、Cloud Run functions 和 Pub/Sub API。

启用 API 所需的角色

如需启用 API，您需要拥有 serviceusage.services.enable 权限。如果您 创建了项目，则可能已通过 Owner 角色 (roles/owner) 拥有此权限。否则，您可以通过 Service Usage Admin 角色 (roles/serviceusage.serviceUsageAdmin) 获得此权限。 了解如何授予角色。

启用 API

gcloud

启用 Managed Airflow、Cloud Run functions 和 Pub/Sub API：

启用 API 所需的角色

如需启用 API，您需要拥有 serviceusage.services.enable 权限。如果您 创建了项目，则可能已通过 Owner 角色 (roles/owner) 拥有此权限。否则，您可以通过 Service Usage Admin 角色 (roles/serviceusage.serviceUsageAdmin) 获得此权限。 了解如何授予角色。

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

Terraform

如需在项目中启用 Cloud Composer API，请将以下资源定义添加到 Terraform 脚本中：

resource "google_project_service" "composer_api" {
  project = "<PROJECT_ID>"
  service = "composer.googleapis.com"
  // Disabling Cloud Composer API might irreversibly break all other
  // environments in your project.
  // This parameter prevents automatic disabling
  // of the API when the resource is destroyed.
  // We recommend to disable the API only after all environments are deleted.
  disable_on_destroy = false
// this flag is introduced in 5.39.0 version of Terraform. If set to true it will
//prevent you from disabling composer_api through Terraform if any environment was
//there in the last 30 days
  check_if_service_has_usage_on_destroy = true
}

resource "google_project_service" "pubsub_api" {
  project = "<PROJECT_ID>"
  service = "pubsub.googleapis.com"
  disable_on_destroy = false
}

resource "google_project_service" "functions_api" {
  project = "<PROJECT_ID>"
  service = "cloudfunctions.googleapis.com"
  disable_on_destroy = false
}

将 <PROJECT_ID> 替换为 项目 ID 您的项目。例如，example-project。

创建 Managed Airflow 环境

创建 Managed Airflow（第 2 代）环境。

在此过程中， 您将 Cloud Composer v2 API Service Agent Extension （roles/composer.ServiceAgentV2Ext）角色授予 Composer Service Agent 账号。Managed Airflow 使用此账号在您的项目中执行操作 in your Google Cloud project。

创建 Pub/Sub 主题

此示例会触发 DAG 以响应推送到 Pub/Sub 主题的消息。创建 Pub/Sub 主题以在此示例中使用：

控制台

1. 在 Google Cloud 控制台中，前往 Pub/Sub 主题 页面。

   前往“Pub/Sub 主题”

2. 点击创建主题 。

3. 在主题 ID 字段中，输入 dag-topic-trigger 作为主题的 ID。

4. 将其他选项保留为默认值。

5. 点击创建主题 。

gcloud

如需创建主题，请在 Google Cloud CLI 中运行 gcloud pubsub topics create 命令：

gcloud pubsub topics create dag-topic-trigger

Terraform

将以下资源定义添加到 Terraform 脚本中：

resource "google_pubsub_topic" "trigger" {
  project                    = "<PROJECT_ID>"
  name                       = "dag-topic-trigger"
  message_retention_duration = "86600s"
}

将 <PROJECT_ID> 替换为 项目 ID 您的项目。例如，example-project。

上传 DAG

将 DAG 上传到您的环境：

1. 将以下 DAG 文件保存在本地计算机上。
2. 将 <PROJECT_ID> 替换为 项目 ID 您的项目。例如，example-project。
3. 将经过修改的 DAG 文件上传到您的环境。

from __future__ import annotations

from datetime import datetime
import time

from airflow import DAG
from airflow import XComArg
from airflow.operators.python import PythonOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.providers.google.cloud.operators.pubsub import (
    PubSubCreateSubscriptionOperator,
    PubSubPullOperator,
)

PROJECT_ID = "<PROJECT_ID>"
TOPIC_ID = "dag-topic-trigger"
SUBSCRIPTION = "trigger_dag_subscription"


def handle_messages(pulled_messages, context):
    dag_ids = list()
    for idx, m in enumerate(pulled_messages):
        data = m.message.data.decode("utf-8")
        print(f"message {idx} data is {data}")
        dag_ids.append(data)
    return dag_ids


# This DAG will run minutely and handle pub/sub messages by triggering target DAG
with DAG(
    "trigger_dag",
    start_date=datetime(2021, 1, 1),
    schedule_interval="* * * * *",
    max_active_runs=1,
    catchup=False,
) as trigger_dag:
    # If subscription exists, we will use it. If not - create new one
    subscribe_task = PubSubCreateSubscriptionOperator(
        task_id="subscribe_task",
        project_id=PROJECT_ID,
        topic=TOPIC_ID,
        subscription=SUBSCRIPTION,
    )

    subscription = subscribe_task.output

    # Proceed maximum 50 messages in callback function handle_messages
    # Here we acknowledge messages automatically. You can use PubSubHook.acknowledge to acknowledge in downstream tasks
    # https://airflow.apache.org/docs/apache-airflow-providers-google/stable/_api/airflow/providers/google/cloud/hooks/pubsub/index.html#airflow.providers.google.cloud.hooks.pubsub.PubSubHook.acknowledge
    pull_messages_operator = PubSubPullOperator(
        task_id="pull_messages_operator",
        project_id=PROJECT_ID,
        ack_messages=True,
        messages_callback=handle_messages,
        subscription=subscription,
        max_messages=50,
    )

    # Here we use Dynamic Task Mapping to trigger DAGs according to messages content
    # https://airflow.apache.org/docs/apache-airflow/2.3.0/concepts/dynamic-task-mapping.html
    trigger_target_dag = TriggerDagRunOperator.partial(task_id="trigger_target").expand(
        trigger_dag_id=XComArg(pull_messages_operator)
    )

    (subscribe_task >> pull_messages_operator >> trigger_target_dag)


def _some_heavy_task():
    print("Do some operation...")
    time.sleep(1)
    print("Done!")


# Simple target DAG
with DAG(
    "target_dag",
    start_date=datetime(2022, 1, 1),
    # Not scheduled, trigger only
    schedule_interval=None,
    catchup=False,
) as target_dag:
    some_heavy_task = PythonOperator(
        task_id="some_heavy_task", python_callable=_some_heavy_task
    )

    (some_heavy_task)

示例代码包含两个 DAG：trigger_dag 和 target_dag。

trigger_dag DAG 会订阅 Pub/Sub 主题，拉取 Pub/Sub 消息，并触发 Pub/Sub 消息数据的 DAG ID 中指定的另一个 DAG。在此示例中，trigger_dag 会触发 target_dag DAG，该 DAG 会将消息输出到任务日志。

trigger_dag DAG 包含以下任务：

• subscribe_task：订阅 Pub/Sub 主题。
• pull_messages_operator：使用 PubSubPullOperator 读取 Pub/Sub 消息数据。
• trigger_target_dag：根据从 Pub/Sub 主题拉取的消息中的数据触发另一个 DAG（在此示例中为 target_dag）。

target_dag DAG 仅包含一项任务：output_to_logs。此任务会以一秒的延迟将消息打印到任务日志。

注意： 如果您想将更多参数（例如 conf 或 execution_date）传递给 DAG 运行，则可以修改 Cloud Run 函数和 DAG 示例。在 Pub/Sub 消息中传递值，然后在 trigger_target_dag 任务中处理这些值。如需详细了解可以传递的参数，请参阅 TriggerDagRunOperator的文档。

部署在 Pub/Sub 主题上发布消息的 Cloud Run 函数

在本部分中，您将部署一个在 Pub/Sub 主题上发布消息的 Cloud Run 函数。

创建 Cloud Run 函数并指定其配置

控制台

1. 在 Google Cloud 控制台中，前往 Cloud Run functions 页面。

   前往 Cloud Run functions

2. 点击创建函数 。

3. 在环境 字段中，选择第 1 代 。

4. 在函数名称 字段中，输入函数的名称：pubsub-publisher。

5. 在触发器类型 字段中，选择 HTTP 。

6. 在身份验证 部分中，选择允许未通过身份验证的调用 。此选项允许 未经身份验证的用户 调用 HTTP 函数。

   注意： 对于生产用途，请考虑使用需要 身份验证的函数。如需了解详情，请参阅 进行身份验证以便调用 。

7. 点击保存。

8. 点击下一步 以继续执行代码 步骤。

Terraform

请考虑在此步骤中使用 Google Cloud 控制台，因为没有 直接从 Terraform 管理函数源代码的方法。

此示例演示了如何通过创建 Cloud Storage 存储桶、将文件存储在此存储桶中，然后使用存储桶中的文件作为 Cloud Run 函数的来源，从本地 zip 归档文件上传 Cloud Run 函数。如果您使用此方法，即使您创建新的归档文件，Terraform 也不会自动更新函数的源代码。如需重新上传函数代码，您可以更改归档文件的文件名。

1. 下载 pubsub_publisher.py 和 requirements.txt 文件。
2. 在 pubsub_publisher.py 文件中，将 <PROJECT_ID> 替换为您的项目的 项目 ID。例如，example-project。
3. 使用 pbusub_publisner.py 和 requirements.txt 文件创建一个名为 pubsub_function.zip 的 zip 归档文件。
4. 将 zip 归档文件保存到存储 Terraform 脚本的目录。
5. 将以下资源定义添加到 Terraform 脚本中，并将 <PROJECT_ID> 替换为您的项目的项目 ID。

resource "google_storage_bucket" "cloud_function_bucket" {
  project        = <PROJECT_ID>
  name           = "<PROJECT_ID>-cloud-function-source-code"
  location       = "US"
  force_destroy  = true
  uniform_bucket_level_access = true
}

resource "google_storage_bucket_object" "cloud_function_source" {
  name   = "pubsub_function.zip"
  bucket = google_storage_bucket.cloud_function_bucket.name
  source = "./pubsub_function.zip"
}

resource "google_cloudfunctions_function" "pubsub_function" {
  project = <PROJECT_ID>
  name    = "pubsub-publisher"
  runtime = "python310"
  region  = "us-central1"

  available_memory_mb   = 128
  source_archive_bucket = google_storage_bucket.cloud_function_bucket.name
  source_archive_object = "pubsub_function.zip"
  timeout               = 60
  entry_point           = "pubsub_publisher"
  trigger_http          = true
}

指定 Cloud Run 函数代码参数

控制台

1. 在代码 步骤的运行时 字段中，选择函数要使用的语言运行时。在此示例中，选择 Python 3.10 。

2. 在入口点 字段中，输入 pubsub_publisher。这是 Cloud Run 函数运行时执行的代码。此标志的值必须是源代码中存在的函数名称或完全限定类名称。

Terraform

跳过此步骤。Cloud Run 函数参数已在 google_cloudfunctions_function 资源中定义。

上传 Cloud Run 函数代码

控制台

在源代码字段中，选择有关如何 提供函数源代码的适当选项。 在本教程中，请使用 Cloud Run functions 内嵌编辑器 添加函数代码。或者，您也可以上传 ZIP 文件，或使用 Cloud Source Repositories。

1. 将以下代码示例放入 main.py 文件中。
2. 将 <PROJECT_ID> 替换为 项目 ID 您的项目。例如，example-project。

from google.cloud import pubsub_v1

project = "<PROJECT_ID>"
topic = "dag-topic-trigger"


def pubsub_publisher(request):
    """Publish message from HTTP request to Pub/Sub topic.
    Args:
        request (flask.Request): HTTP request object.
    Returns:
        The response text with message published into Pub/Sub topic
        Response object using
        `make_response <http://flask.pocoo.org/docs/1.0/api/#flask.Flask.make_response>`.
    """
    request_json = request.get_json()
    print(request_json)
    if request.args and "message" in request.args:
        data_str = request.args.get("message")
    elif request_json and "message" in request_json:
        data_str = request_json["message"]
    else:
        return "Message content not found! Use 'message' key to specify"

    publisher = pubsub_v1.PublisherClient()
    # The `topic_path` method creates a fully qualified identifier
    # in the form `projects/{project_id}/topics/{topic_id}`
    topic_path = publisher.topic_path(project, topic)

    # The required data format is a bytestring
    data = data_str.encode("utf-8")
    # When you publish a message, the client returns a future.
    message_length = len(data_str)
    future = publisher.publish(topic_path, data, message_length=str(message_length))
    print(future.result())

    return f"Message {data} with message_length {message_length} published to {topic_path}."

Terraform

跳过此步骤。Cloud Run 函数参数已在 google_cloudfunctions_function 资源中定义。

指定 Cloud Run 函数依赖项

控制台

在 requirements.txt 元数据文件中指定函数依赖项：

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

部署函数时，Cloud Run functions 会下载并安装 requirements.txt 文件中声明的依赖项，每个软件包占一行。 此文件必须位于包含您的函数代码的 main.py 文件所在的目录中。如需了解详情，请参阅 Requirements Files in pip 文档。

Terraform

跳过此步骤。Cloud Run 函数依赖项在 pubsub_function.zip 归档文件的 requirements.txt 文件中定义。

部署 Cloud Run 函数

控制台

点击部署 。部署成功完成后，函数会显示 在控制台的 Cloud Run functions 页面中，并带有绿色对勾标记 Google Cloud 。

确保运行 Cloud Run 函数的服务帐号 在您的项目中拥有足够的权限来访问 Pub/Sub。

Terraform

1. 初始化 Terraform：

   terraform init
   

2. 查看配置并验证 Terraform 将创建或更新的资源是否符合您的预期：

   terraform plan
   

3. 如需检查配置是否有效，请运行以下命令：

   terraform validate
   

4. 通过运行以下命令并在提示符处输入 yes 来应用 Terraform 配置：

   terraform apply
   

等待 Terraform 显示“应用完成！”消息。

在 Google Cloud 控制台的界面中找到资源，以确保 Terraform 已创建或更新它们。

测试 Cloud Run 函数

如需检查函数是否在 Pub/Sub 主题上发布消息，以及示例 DAG 是否按预期运行，请执行以下操作：

1. 检查 DAG 是否处于活跃状态：

   1. 在 Google Cloud 控制台中，前往环境页面。

      前往“环境”

   2. 在环境列表中，点击您的环境名称。环境详情 页面会打开。

   3. 前往 DAG 标签页。

   4. 检查名为 trigger_dag 和 target_dag 的 DAG 的状态 列中的值。这两个 DAG 都必须处于 Active 状态。

2. 推送测试 Pub/Sub 消息。您可以在 Cloud Shell 中执行此操作：

   1. 在 Google Cloud 控制台中，前往 函数 页面。

      前往 Cloud Run functions

   2. 点击函数的名称 pubsub-publisher。

   3. 前往测试 标签页。

   4. 在配置触发事件 部分中，输入以下 JSON 键值对：{"message": "target_dag"}。请勿修改键值对，因为此消息稍后会触发测试 DAG。

   5. 在测试命令 部分中，点击在 Cloud Shell 中测试 。

   6. 在 Cloud Shell 终端 中，等待命令自动显示 。按 Enter 运行此命令。

   7. 如果显示为 Cloud Shell 提供授权 消息，请点击授权 。

   8. 检查消息内容是否与 Pub/Sub 消息对应。在此示例中，输出消息必须以 Message b'target_dag' with message_length 10 published to开头，作为 函数的响应。

3. 检查 target_dag 是否已触发：

   1. 等待至少一分钟，以便 trigger_dag 的新 DAG 运行完成。

   2. 在 Google Cloud 控制台中，前往环境页面。

      前往“环境”

   3. 在环境列表中，点击您的环境名称。环境详情 页面会打开。

   4. 前往 DAG 标签页。

   5. 点击 trigger_dag 以前往 DAG 详情 页面。在运行 标签页上，系统会显示 trigger_dag DAG 的 DAG 运行列表。

      此 DAG 每分钟运行一次，并处理从函数发送的所有 Pub/Sub 消息。如果未发送任何消息，则 trigger_target 任务在 DAG 运行日志中标记为 Skipped。如果触发了 DAG，则 trigger_target 任务标记为 Success。

   6. 浏览几个最近的 DAG 运行，找到一个 DAG 运行，其中所有三个任务（subscribe_task、pull_messages_operator 和 trigger_target）的状态均为 Success。

   7. 返回到 DAG 标签页，并检查 target_dag DAG 的成功运行次数 列是否列出一次成功运行。

摘要

在本教程中，您学习了如何使用 Cloud Run functions 在 Pub/Sub 主题上发布消息，以及如何部署订阅 Pub/Sub 主题、拉取 Pub/Sub 消息并触发消息数据的 DAG ID 中指定的另一个 DAG 的 DAG。

此外，还有其他 创建和管理 Pub/Sub 订阅 以及 触发 DAG 的方法，这些方法不在本教程的讨论范围内。例如，您可以 使用 Cloud Run functions 在发生指定事件时触发 Airflow DAG 。 请查看我们的教程，自行试用其他 Google Cloud 功能。

清理

为避免因本教程中使用的资源导致您的 Google Cloud 账号产生费用，请删除包含这些资源的 项目，或者保留该项目但删除各个资源。

删除项目

警告：删除项目会产生以下影响

• 项目中的所有内容都会被删除。如果您将现有项目用于本文档中的任务，则删除该项目后，还将删除您已在该项目中完成的任何其他工作。
• 自定义项目 ID 丢失。创建此项目时，您可能创建了要在将来使用的自定义项目 ID。要保留使用该项目 ID 的网址（例如 appspot.com 网址），请删除项目内的所选资源，而不是删除整个项目。

如果您打算探索多个架构、教程或快速入门，则重复使用项目 可以帮助您避免超出项目配额限制。

删除 Google Cloud 项目：

gcloud projects delete PROJECT_ID

删除各个资源

如果您打算探索多个教程和快速入门，重复使用项目可以帮助您避免超出项目配额上限。

控制台

1. 删除 Managed Airflow 环境。在此过程中，您还会删除环境的存储桶。
2. 删除 Pub/Sub 主题，dag-topic-trigger。

3. 删除 Cloud Run 函数。

   1. 在 Google Cloud 控制台中，前往 Cloud Run functions。

      前往 Cloud Run functions

   2. 点击要删除的函数 pubsub-publisher 对应的复选框。

   3. 点击删除，然后按照说明操作。

Terraform

1. 确保 Terraform 脚本不包含项目中仍需要的资源的条目。例如，您可能希望保留一些已启用的 API，并仍分配 IAM 权限（如果您已将此类定义添加到 Terraform 脚本中）。
2. 运行 terraform destroy。
3. 手动删除环境的存储桶。Managed Airflow 不会自动删除它。您可以从 控制台 Google Cloud 或 Google Cloud CLI 执行此操作。

后续步骤

• 测试 DAG
• 测试 HTTP 函数
• 部署 Cloud Run 函数
• 自行试用其他 Google Cloud 功能。查看 我们的教程。