Memicu DAG menggunakan Cloud Functions dan Pesan Pub/Sub

Cloud Composer 1 | Cloud Composer 2

Halaman ini akan memandu Anda dalam membuat arsitektur push berbasis peristiwa dengan memicu DAG Cloud Composer sebagai respons terhadap perubahan topik Pub/Sub. Contoh dalam tutorial ini menunjukkan penanganan siklus penuh pengelolaan Pub/Sub, termasuk pengelolaan langganan, sebagai bagian dari proses DAG. Metode ini cocok untuk beberapa kasus penggunaan umum saat Anda perlu memicu DAG, tetapi tidak ingin menyiapkan izin akses tambahan.

Misalnya, pesan yang dikirim melalui Pub/Sub dapat digunakan sebagai solusi jika Anda tidak ingin memberikan akses langsung ke lingkungan Cloud Composer karena alasan keamanan. Anda dapat mengonfigurasi Cloud Function yang membuat pesan Pub/Sub dan memublikasikannya berdasarkan topik Pub/Sub. Selanjutnya, Anda dapat membuat DAG yang menarik pesan Pub/Sub, lalu menangani pesan ini.

Dalam contoh spesifik ini, Anda membuat Cloud Function dan men-deploy dua DAG. DAG pertama menarik pesan Pub/Sub dan memicu DAG kedua sesuai dengan konten pesan Pub/Sub.

Tutorial ini mengasumsikan bahwa Anda telah memahami Python dan konsol Google Cloud.

Tujuan

Biaya

Tutorial ini menggunakan komponen Google Cloud yang dapat ditagih berikut:

• Cloud Composer (lihat juga biaya tambahan)
• Pub/Sub
• Cloud Functions

Setelah menyelesaikan tutorial ini, Anda dapat menghindari penagihan berkelanjutan dengan menghapus resource yang Anda buat. Lihat Pembersihan untuk mengetahui detail selengkapnya.

Sebelum memulai

Untuk tutorial ini, Anda memerlukan project Google Cloud. Konfigurasikan project dengan cara berikut:

1. Di konsol Google Cloud, pilih atau buat project:

   Buka Project Selector

2. Pastikan penagihan diaktifkan untuk project Anda. Pelajari cara memeriksa apakah penagihan diaktifkan di sebuah project.

3. Pastikan pengguna project Google Cloud Anda memiliki peran berikut untuk membuat resource yang diperlukan:

   • Service Account User (roles/iam.serviceAccountUser)
   • Pub/Sub Editor (roles/pubsub.editor)
   • Administrator Objek Lingkungan dan Penyimpanan (roles/composer.environmentAndStorageObjectAdmin)
   • Admin Cloud Functions (roles/cloudfunctions.admin)
   • Viewer Log (roles/logging.viewer)

4. Pastikan akun layanan yang menjalankan Cloud Function Anda memiliki izin yang memadai dalam project Anda untuk mengakses Pub/Sub. Secara default, Cloud Functions menggunakan akun layanan default App Engine. Akun layanan ini memiliki peran Editor, yang memiliki izin memadai untuk tutorial ini.

Mengaktifkan API untuk project Anda

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
}

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
}

Membuat lingkungan Cloud Composer

Buat lingkungan Cloud Composer 2.

Sebagai bagian dari prosedur ini, Anda memberikan peran Cloud Composer v2 API Service Agent Extension (roles/composer.ServiceAgentV2Ext) ke akun Agen Layanan Composer. Cloud Composer menggunakan akun ini untuk menjalankan operasi di project Google Cloud Anda.

Membuat topik Pub/Sub

Contoh ini memicu DAG sebagai respons terhadap pesan yang dikirim ke topik Pub/Sub. Buat topik Pub/Sub untuk digunakan dalam contoh ini:

gcloud pubsub topics create dag-topic-trigger

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

Mengupload DAG Anda

Upload DAG ke lingkungan Anda:

1. Simpan file DAG berikut di komputer lokal Anda.
2. Ganti <PROJECT_ID> dengan Project ID dari project Anda. Misalnya, example-project.
3. Upload file DAG yang telah diedit ke lingkungan Anda.

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)

Kode contoh berisi dua DAG: trigger_dag dan target_dag.

DAG trigger_dag berlangganan topik Pub/Sub, menarik pesan Pub/Sub, dan memicu DAG lain yang ditentukan dalam ID DAG dari data pesan Pub/Sub. Dalam contoh ini, trigger_dag memicu DAG target_dag, yang menghasilkan pesan ke log tugas.

DAG trigger_dag berisi tugas-tugas berikut:

• subscribe_task: Berlangganan topik Pub/Sub.
• pull_messages_operator: Membaca data pesan Pub/Sub dengan PubSubPullOperator.
• trigger_target_dag: Memicu DAG lain (dalam contoh ini, target_dag) sesuai dengan data dalam pesan yang diambil dari topik Pub/Sub.

DAG target_dag hanya berisi satu tugas: output_to_logs. Tugas ini mencetak pesan ke log tugas dengan penundaan satu detik.

Men-deploy Cloud Function yang memublikasikan pesan tentang topik Pub/Sub

Di bagian ini, Anda akan men-deploy Cloud Function yang memublikasikan pesan tentang topik Pub/Sub.

Membuat Cloud Function dan menentukan konfigurasinya

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
}

Menentukan parameter kode Cloud Function

Upload kode Cloud Function Anda

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

Menentukan dependensi Cloud Function

requests-toolbelt==1.0.0
google-auth==2.19.1
google-cloud-pubsub==2.17.0

Men-deploy Cloud Function

Menguji Cloud Function Anda

Untuk memeriksa apakah fungsi Anda memublikasikan pesan di topik Pub/Sub dan contoh DAG berfungsi sebagaimana mestinya:

1. Pastikan DAG sudah aktif:

   1. Di konsol Google Cloud, buka halaman Environments.

      Buka Lingkungan

   2. Pada daftar lingkungan, klik nama lingkungan Anda. Halaman Detail lingkungan akan terbuka.

   3. Buka tab DAGs.

   4. Periksa nilai di kolom State untuk DAG bernama trigger_dag dan target_dag. Kedua DAG harus dalam status Active.

2. Kirim pesan Pub/Sub pengujian. Anda dapat melakukannya di Cloud Shell:

   1. Di konsol Google Cloud, buka halaman Functions.

      Buka Cloud Functions

   2. Klik nama fungsi Anda, pubsub-publisher.

   3. Buka tab Testing.

   4. Di bagian Mengonfigurasi peristiwa pemicu, masukkan nilai kunci JSON berikut: {"message": "target_dag"}. Jangan ubah key-value pair karena pesan ini akan memicu DAG pengujian.

   5. Di bagian Test Command, klik Test in Cloud Shell.

   6. Di Terminal Cloud Shell, tunggu hingga perintah muncul secara otomatis. Jalankan perintah ini dengan menekan Enter.

   7. Jika pesan Authorize Cloud Shell muncul, klik Authorize.

   8. Periksa apakah konten pesan sesuai dengan pesan Pub/Sub. Dalam contoh ini, pesan output harus diawali dengan Message b'target_dag' with message_length 10 published to sebagai respons dari fungsi Anda.

3. Pastikan target_dag dipicu:

   1. Tunggu setidaknya satu menit, sehingga operasi DAG baru trigger_dag selesai.

   2. Di konsol Google Cloud, buka halaman Environments.

      Buka Lingkungan

   3. Pada daftar lingkungan, klik nama lingkungan Anda. Halaman Detail lingkungan akan terbuka.

   4. Buka tab DAGs.

   5. Klik trigger_dag untuk membuka halaman Detail DAG. Di tab Runs, akan ditampilkan daftar DAG yang dijalankan untuk DAG trigger_dag.

      DAG ini berjalan setiap menit dan memproses semua pesan Pub/Sub yang dikirim dari fungsi. Jika tidak ada pesan yang dikirim, tugas trigger_target akan ditandai sebagai Skipped di log pengoperasian DAG. Jika DAG dipicu, tugas trigger_target akan ditandai sebagai Success.

   6. Lihat beberapa operasi DAG terbaru untuk menemukan proses DAG dengan ketiga tugas (subscribe_task, pull_messages_operator, dan trigger_target) memiliki status Success.

   7. Kembali ke tab DAG dan periksa apakah kolom Successful run untuk DAG target_dag mencantumkan satu operasi yang berhasil.

Ringkasan

Dalam tutorial ini, Anda telah mempelajari cara menggunakan Cloud Functions untuk memublikasikan pesan pada topik Pub/Sub dan men-deploy DAG yang berlangganan topik Pub/Sub, menarik pesan Pub/Sub, dan memicu DAG lain yang ditentukan dalam ID DAG data pesan.

Ada juga cara alternatif untuk membuat dan mengelola langganan Pub/Sub serta memicu DAG yang berada di luar cakupan tutorial ini. Misalnya, Anda dapat menggunakan Cloud Functions untuk memicu DAG Airflow saat peristiwa tertentu terjadi. Lihat tutorial kami untuk mencoba sendiri fitur Google Cloud.

Pembersihan

Agar tidak menimbulkan biaya pada akun Google Cloud Anda untuk resource yang digunakan dalam tutorial ini, hapus project yang berisi resource atau simpan project tersebut dan hapus resource individual.

Menghapus project

Menghapus project Google Cloud:

gcloud projects delete PROJECT_ID

Menghapus resource satu per satu

Jika Anda berencana mempelajari beberapa tutorial dan panduan memulai, menggunakan kembali project dapat membantu Anda agar tidak melampaui batas kuota project.

Langkah selanjutnya

• DAG Pengujian
• Menguji Fungsi HTTP
• Men-deploy Cloud Function
• Coba sendiri fitur Google Cloud lainnya. Lihat tutorial kami.