DAGs mit Cloud Functions und Pub/Sub-Nachrichten auslösen

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Auf dieser Seite erfahren Sie, wie Sie eine ereignisbasierte Push-Architektur erstellen, indem Sie Cloud Composer-DAGs als Reaktion auf Änderungen an Pub/Sub-Themen auslösen. Die Beispiele in dieser Anleitung zeigen, wie der gesamte Zyklus der Pub/Sub-Verwaltung, einschließlich der Aboverwaltung, als Teil des DAG-Prozesses behandelt wird. Sie eignet sich für einige der häufigsten Anwendungsfälle, in denen Sie DAGs auslösen müssen, aber keine zusätzlichen Zugriffsberechtigungen einrichten möchten.

Nachrichten, die über Pub/Sub gesendet werden, können beispielsweise als Lösung verwendet werden, wenn Sie aus Sicherheitsgründen keinen direkten Zugriff auf eine Cloud Composer-Umgebung gewähren möchten. Sie können eine Cloud Run-Funktion konfigurieren, die Pub/Sub-Nachrichten erstellt und in einem Pub/Sub-Thema veröffentlicht. Anschließend können Sie einen DAG erstellen, der Pub/Sub-Nachrichten abruft und verarbeitet.

In diesem Beispiel erstellen Sie eine Cloud Run-Funktion und stellen zwei DAGs bereit. Der erste DAG ruft Pub/Sub-Nachrichten ab und löst den zweiten DAG entsprechend dem Inhalt der Pub/Sub-Nachricht aus.

In dieser Anleitung wird davon ausgegangen, dass Sie mit Python und der Google Cloud Konsole vertraut sind.

Lernziele

Kosten

In dieser Anleitung werden die folgenden kostenpflichtigen Komponenten von Google Cloudverwendet:

• Cloud Composer (siehe auch zusätzliche Kosten)
• Pub/Sub
• Cloud Run Functions

Nach Abschluss dieser Anleitung können Sie weitere Kosten vermeiden, indem Sie die erstellten Ressourcen löschen. Weitere Informationen finden Sie unter Bereinigen.

Hinweise

Für diese Anleitung benötigen Sie ein Google Cloud-Projekt. Konfigurieren Sie das Projekt so:

1. Wählen Sie in der Google Cloud -Console ein Projekt aus oder erstellen Sie ein Projekt:

   Zur Projektauswahl

2. Die Abrechnung für Ihr Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist.

3. Achten Sie darauf, dass der Nutzer Ihres Google Cloud Projekts die folgenden Rollen hat, um die erforderlichen Ressourcen zu erstellen:

   • Dienstkontonutzer (roles/iam.serviceAccountUser)
   • Pub/Sub-Bearbeiter (roles/pubsub.editor)
   • Administrator für Umgebung und Storage-Objekte (roles/composer.environmentAndStorageObjectAdmin)
   • Cloud Run Functions-Administrator (roles/cloudfunctions.admin)
   • Loganzeige (roles/logging.viewer)

4. Achten Sie darauf, dass das Dienstkonto, mit dem Ihre Cloud Run-Funktion ausgeführt wird, in Ihrem Projekt ausreichende Berechtigungen für den Zugriff auf Pub/Sub hat. Standardmäßig verwenden Cloud Run-Funktionen das App Engine-Standarddienstkonto. Dieses Dienstkonto hat die Rolle Bearbeiter, die für diese Anleitung ausreichende Berechtigungen bietet.

Die APIs für Ihr Projekt aktivieren

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

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
}

Cloud Composer-Umgebung erstellen

Cloud Composer 2-Umgebung erstellen

Im Rahmen dieser Vorgehensweise weisen Sie dem Composer-Dienst-Agent-Konto die Rolle Dienst-Agent-Erweiterung für die Cloud Composer v2 API (roles/composer.ServiceAgentV2Ext) zu. Cloud Composer verwendet dieses Konto, um Vorgänge in Ihrem Google Cloud -Projekt auszuführen.

Pub/Sub-Thema erstellen

In diesem Beispiel wird ein DAG als Reaktion auf eine Nachricht ausgelöst, die per Push an ein Pub/Sub-Thema übertragen wird. Erstellen Sie ein Pub/Sub-Thema, das Sie in diesem Beispiel verwenden können:

gcloud pubsub topics create dag-topic-trigger

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

DAGs hochladen

So laden Sie DAGs in Ihre Umgebung hoch:

1. Speichern Sie die folgende DAG-Datei auf Ihrem lokalen Computer.
2. Ersetzen Sie <PROJECT_ID> durch die Projekt-ID Ihres Projekts. Beispiel: example-project
3. Laden Sie die bearbeitete DAG-Datei in Ihre Umgebung hoch.

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)

Der Beispielcode enthält zwei DAGs: trigger_dag und target_dag.

Der DAG trigger_dag abonniert ein Pub/Sub-Thema, ruft Pub/Sub-Nachrichten ab und löst einen anderen DAG aus, der in der DAG-ID der Pub/Sub-Nachrichtendaten angegeben ist. In diesem Beispiel löst trigger_dag den DAG target_dag aus, der Nachrichten in die Aufgabenlogs ausgibt.

Der trigger_dag-DAG enthält die folgenden Aufgaben:

• subscribe_task: Abonnieren Sie ein Pub/Sub-Thema.
• pull_messages_operator: Pub/Sub-Nachrichtendaten mit PubSubPullOperator lesen.
• trigger_target_dag: Eine andere DAG (in diesem Beispiel target_dag) wird gemäß den Daten in den Nachrichten ausgelöst, die aus dem Pub/Sub-Thema abgerufen werden.

Der target_dag-DAG enthält nur eine Aufgabe: output_to_logs. Bei dieser Aufgabe werden Nachrichten mit einer Sekunde Verzögerung in das Aufgabenprotokoll ausgegeben.

Cloud Run-Funktion bereitstellen, die Nachrichten in einem Pub/Sub-Thema veröffentlicht

In diesem Abschnitt stellen Sie eine Cloud Run-Funktion bereit, die Nachrichten in einem Pub/Sub-Thema veröffentlicht.

Cloud Run-Funktion erstellen und Konfiguration angeben

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-Funktions-Codeparameter angeben

Cloud Run Functions-Code hochladen

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}."

Abhängigkeiten für Cloud Run Functions angeben

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

Cloud Run-Funktion bereitstellen

Cloud Run-Funktion testen

So prüfen Sie, ob Ihre Funktion eine Nachricht in einem Pub/Sub-Thema veröffentlicht und die Beispiel-DAGs wie vorgesehen funktionieren:

1. Prüfen Sie, ob die DAGs aktiv sind:

   1. Rufen Sie in der Google Cloud -Console die Seite Umgebungen auf.

      Zur Seite Umgebungen

   2. Klicken Sie in der Liste der Umgebungen auf den Namen Ihrer Umgebung. Die Seite Umgebungsdetails wird geöffnet.

   3. Rufen Sie den Tab DAGs auf.

   4. Prüfen Sie die Werte in der Spalte Status für DAGs mit den Namen trigger_dag und target_dag. Beide DAGs müssen den Status Active haben.

2. Senden Sie eine Pub/Sub-Testnachricht. Sie können dies in Cloud Shell tun:

   1. Wechseln Sie in der Google Cloud Console zur Seite Functions.

      Zu den Cloud Run-Funktionen

   2. Klicken Sie auf den Namen Ihrer Funktion, pubsub-publisher.

   3. Rufen Sie den Tab Testen auf.

   4. Geben Sie im Abschnitt Auslösendes Ereignis konfigurieren das folgende JSON-Schlüssel/Wert-Paar ein: {"message": "target_dag"}. Ändern Sie das Schlüssel/Wert-Paar nicht, da diese Nachricht später den Test-DAG auslöst.

   5. Klicken Sie im Bereich Testbefehl auf In Cloud Shell testen.

   6. Warten Sie im Cloud Shell-Terminal, bis ein Befehl automatisch angezeigt wird. Führen Sie diesen Befehl aus, indem Sie Enter drücken.

   7. Wenn die Meldung Cloud Shell autorisieren angezeigt wird, klicken Sie auf Autorisieren.

   8. Prüfen Sie, ob der Nachrichteninhalt der Pub/Sub-Nachricht entspricht. In diesem Beispiel muss die Ausgabenachricht mit Message b'target_dag' with message_length 10 published to als Antwort Ihrer Funktion beginnen.

3. Prüfen Sie, ob target_dag ausgelöst wurde:

   1. Warten Sie mindestens eine Minute, bis ein neuer DAG-Lauf von trigger_dag abgeschlossen ist.

   2. Rufen Sie in der Google Cloud -Console die Seite Umgebungen auf.

      Zur Seite Umgebungen

   3. Klicken Sie in der Liste der Umgebungen auf den Namen Ihrer Umgebung. Die Seite Umgebungsdetails wird geöffnet.

   4. Rufen Sie den Tab DAGs auf.

   5. Klicken Sie auf trigger_dag, um die Seite DAG-Details aufzurufen. Auf dem Tab Ausführungen wird eine Liste der DAG-Ausführungen für die trigger_dag-DAG angezeigt.

      Dieser DAG wird jede Minute ausgeführt und verarbeitet alle Pub/Sub-Nachrichten, die von der Funktion gesendet werden. Wenn keine Nachrichten gesendet wurden, wird die Aufgabe trigger_target in den DAG-Laufprotokollen als Skipped markiert. Wenn DAGs ausgelöst wurden, wird die Aufgabe trigger_target als Success markiert.

   6. Sehen Sie sich mehrere aktuelle DAG-Ausführungen an, um eine DAG-Ausführung zu finden, in der alle drei Aufgaben (subscribe_task, pull_messages_operator und trigger_target) den Status Success haben.

   7. Kehren Sie zum Tab DAGs zurück und prüfen Sie, ob in der Spalte Erfolgreiche Ausführungen für den DAG target_dag eine erfolgreiche Ausführung aufgeführt ist.

Zusammenfassung

In dieser Anleitung haben Sie gelernt, wie Sie Cloud Run Functions verwenden, um Nachrichten in einem Pub/Sub-Thema zu veröffentlichen und einen DAG bereitzustellen, der ein Pub/Sub-Thema abonniert, Pub/Sub-Nachrichten abruft und einen anderen DAG auslöst, der in der DAG-ID der Nachrichtendaten angegeben ist.

Es gibt auch alternative Möglichkeiten zum Erstellen und Verwalten von Pub/Sub-Abos und zum Auslösen von DAGs, die in dieser Anleitung nicht behandelt werden. Sie können beispielsweise Cloud Run-Funktionen verwenden, um Airflow-DAGs auszulösen, wenn ein bestimmtes Ereignis eintritt. Anleitungen für die anderenGoogle Cloud -Funktionen

Bereinigen

Damit Ihrem Google Cloud -Konto die in dieser Anleitung verwendeten Ressourcen nicht in Rechnung gestellt werden, können Sie entweder das Projekt löschen, das die Ressourcen enthält, oder das Projekt beibehalten und die einzelnen Ressourcen löschen.

Projekt löschen

Delete a Google Cloud project:

gcloud projects delete PROJECT_ID

Einzelne Ressourcen löschen

Wenn Sie mehrere Anleitungen und Kurzanleitungen durcharbeiten möchten, können Sie die Überschreitung von Projektkontingenten verhindern, indem Sie Projekte wiederverwenden.

Nächste Schritte

• DAGs testen
• HTTP-Funktionen testen
• Cloud Run-Funktion bereitstellen
• Weitere Google Cloud Funktionen ausprobieren Anleitungen