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:

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

Konsol

Enable the Cloud Composer, Cloud Functions, and Pub/Sub APIs.

Enable the APIs

gcloud

Enable the Cloud Composer, Cloud Functions, and Pub/Sub APIs: 

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

Terraform

Aktifkan Cloud Composer API di project Anda dengan menambahkan definisi resource berikut ke skrip Terraform Anda:

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
}

Ganti <PROJECT_ID> dengan Project ID dari project Anda. Misalnya, example-project.

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:

Konsol

  1. Di konsol Google Cloud, buka halaman Pub/Sub Topics.

    Buka Topik Pub/Sub

  2. Klik Buat Topik.

  3. Di kolom Topic ID, masukkan dag-topic-trigger sebagai ID untuk topik Anda.

  4. Tetap gunakan nilai default untuk opsi lain.

  5. Klik Buat Topik.

gcloud

Untuk membuat topik, jalankan perintah gcloud pubsub topics create di Google Cloud CLI:

gcloud pubsub topics create dag-topic-trigger

Terraform

Tambahkan definisi resource berikut ke skrip Terraform Anda:

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

Ganti <PROJECT_ID> dengan Project ID dari project Anda. Misalnya, example-project.

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

Konsol

  1. Di konsol Google Cloud, buka halaman Cloud Function.

    Buka Cloud Functions

  2. Klik Create function.

  3. Di kolom Environment, pilih 1st gen.

  4. Di kolom Function name, masukkan nama untuk fungsi Anda: pubsub-publisher.

  5. Pada kolom Jenis pemicu, pilih HTTP.

  6. Di bagian Authentication, pilih Allow unauthenticated invocations. Opsi ini memberi pengguna yang tidak diautentikasi kemampuan untuk memanggil fungsi HTTP.

  7. Klik Simpan.

  8. Klik Berikutnya untuk berpindah ke langkah Kode.

Terraform

Pertimbangkan untuk menggunakan Konsol Google Cloud pada langkah ini, karena tidak ada cara mudah untuk mengelola kode sumber fungsi dari Terraform.

Contoh ini menunjukkan cara mengupload Cloud Function dari file arsip zip lokal dengan membuat bucket Cloud Storage, menyimpan file di bucket ini, lalu menggunakan file dari bucket sebagai sumber untuk Cloud Function. Jika Anda menggunakan pendekatan ini, Terraform tidak otomatis memperbarui kode sumber fungsi, meskipun Anda membuat file arsip baru. Untuk mengupload ulang kode fungsi, Anda dapat mengubah nama file arsip.

  1. Download file pubsub_publisher.py dan requirements.txt.
  2. Di file pubsub_publisher.py, ganti <PROJECT_ID> dengan Project ID project Anda. Misalnya, example-project.
  3. Buat arsip zip bernama pubsub_function.zip dengan pbusub_publisner.py dan file requirements.txt.
  4. Simpan arsip zip ke direktori tempat skrip Terraform Anda disimpan.
  5. Tambahkan definisi resource berikut ke skrip Terraform Anda dan ganti <PROJECT_ID> dengan Project ID project Anda.
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

Konsol

  1. Pada langkah Code, di kolom Runtime, pilih runtime bahasa yang digunakan oleh fungsi Anda. Dalam contoh ini, pilih Python 3.10.

  2. Di kolom Entry point, masukkan pubsub_publisher. Ini adalah kode yang dijalankan saat Cloud Function Anda berjalan. Nilai flag ini harus berupa nama fungsi atau nama class yang sepenuhnya memenuhi syarat, yang ada di kode sumber Anda.

Terraform

Lewati langkah ini. Parameter Cloud Function sudah ditentukan dalam resource google_cloudfunctions_function.

Upload kode Cloud Function Anda

Konsol

Di kolom Source code, pilih opsi yang sesuai untuk cara memasukkan kode sumber fungsi. Dalam tutorial ini, tambahkan kode fungsi Anda menggunakan Inline Editor Cloud Functions. Sebagai alternatif, Anda dapat mengupload file ZIP atau menggunakan Cloud Source Repositories.

  1. Masukkan contoh kode berikut ke dalam file main.py.
  2. Ganti <PROJECT_ID> dengan Project ID dari project Anda. Misalnya, 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

Lewati langkah ini. Parameter Cloud Function sudah ditentukan dalam resource google_cloudfunctions_function.

Menentukan dependensi Cloud Function

Konsol

Tentukan dependensi fungsi dalam file metadata requirements.txt:

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

Saat Anda men-deploy fungsi, Cloud Functions akan mendownload dan menginstal dependensi yang dideklarasikan dalam file requirements.txt, satu baris per paket. File ini harus berada di direktori yang sama dengan file main.py yang berisi kode fungsi Anda. Untuk mengetahui detail selengkapnya, lihat File Persyaratan dalam dokumentasi pip.

Terraform

Lewati langkah ini. Dependensi Cloud Function ditentukan dalam file requirements.txt di arsip pubsub_function.zip.

Men-deploy Cloud Function

Konsol

Klik Deploy. Setelah deployment berhasil diselesaikan, fungsi akan muncul dengan tanda centang hijau di halaman Cloud Functions pada Google Cloud Console.

Pastikan akun layanan yang menjalankan Cloud Function Anda memiliki izin yang cukup dalam project untuk mengakses Pub/Sub.

Terraform

  1. Lakukan inisialisasi Terraform:

    terraform init

  2. Tinjau konfigurasi dan pastikan bahwa resource yang dibuat atau diperbarui oleh Terraform sesuai dengan ekspektasi Anda:

    terraform plan

  3. Untuk memeriksa apakah konfigurasi Anda valid, jalankan perintah berikut:

    terraform validate

  4. Terapkan konfigurasi Terraform dengan menjalankan perintah berikut dan memasukkan ya pada perintah:

    terraform apply

Tunggu hingga Terraform menampilkan pesan "Apply complete!".

Di Konsol Google Cloud, buka resource Anda di UI untuk memastikan Terraform telah membuat atau memperbaruinya.

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.

Konsol

  1. Menghapus lingkungan Cloud Composer. Anda juga akan menghapus bucket lingkungan selama prosedur ini.
  2. Hapus topik Pub/Sub, dag-topic-trigger.

  3. Hapus Cloud Function.

    1. Di konsol Google Cloud, buka Cloud Functions.

      Buka Cloud Functions

    2. Klik kotak centang untuk fungsi yang ingin Anda hapus, pubsub-publisher.

    3. Klik Hapus, lalu ikuti petunjuknya.

Terraform

  1. Pastikan skrip Terraform Anda tidak berisi entri untuk resource yang masih diperlukan oleh project Anda. Misalnya, Anda mungkin ingin tetap mengaktifkan beberapa API dan izin IAM tetap ditetapkan (jika Anda menambahkan definisi tersebut ke skrip Terraform).
  2. Jalankan terraform destroy.
  3. Hapus bucket lingkungan secara manual. Cloud Composer tidak otomatis menghapusnya. Anda dapat melakukannya dari Google Cloud Console atau Google Cloud CLI.

Langkah selanjutnya