Memicu DAG menggunakan Cloud Functions dan Pesan Pub/Sub

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Halaman ini memandu Anda membuat arsitektur push berbasis peristiwa dengan memicu DAG Cloud Composer sebagai respons terhadap perubahan topik Pub/Sub. Contoh dalam tutorial ini menunjukkan penanganan siklus lengkap pengelolaan Pub/Sub, termasuk pengelolaan langganan, sebagai bagian dari proses DAG. 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 fungsi Cloud Run yang membuat pesan Pub/Sub dan memublikasikannya di topik Pub/Sub. Kemudian, Anda dapat membuat DAG yang mengambil pesan Pub/Sub, lalu menangani pesan ini.

Catatan: Sebagai pendekatan alternatif, Anda dapat memicu DAG menggunakan hanya Airflow REST API, atau Anda mungkin ingin memicu DAG menggunakan fungsi Cloud Run saat peristiwa tertentu terjadi.

Dalam contoh khusus ini, Anda akan membuat fungsi Cloud Run dan men-deploy dua DAG. DAG pertama mengambil pesan Pub/Sub dan memicu DAG kedua sesuai dengan konten pesan Pub/Sub.

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

Tujuan

Tujuan:

• Buat resource terkait dan siapkan lingkungan Anda.
• Upload DAG yang berlangganan topik Pub/Sub.
• Konfigurasikan dan deploy fungsi Cloud Run yang mengirimkan pesan Pub/Sub ke topik Pub/Sub.
• Uji fungsi Anda.

Biaya

Tutorial ini menggunakan komponen Google Cloud yang dapat ditagih berikut:

• Cloud Composer (lihat juga biaya tambahan)
• Pub/Sub
• Cloud Run 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 Pemilih Project

2. Pastikan penagihan diaktifkan untuk project Anda. Pelajari cara memeriksa apakah penagihan diaktifkan pada 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)
   • Environment and Storage Object Administrator (roles/composer.environmentAndStorageObjectAdmin)
   • Cloud Run functions Admin (roles/cloudfunctions.admin)
   • Logs Viewer (roles/logging.viewer)

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

Mengaktifkan API untuk project Anda

Konsol

Enable the Cloud Composer, Cloud Run functions, and Pub/Sub APIs.

Enable the APIs

gcloud

Enable the Cloud Composer, Cloud Run 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:

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
}

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

Membuat lingkungan Cloud Composer

Buat lingkungan Cloud Composer 2.

Sebagai bagian dari prosedur ini, Anda memberikan peran Ekstensi Agen Layanan Cloud Composer v2 API (roles/composer.ServiceAgentV2Ext) ke akun Agen Layanan Composer. Cloud Composer menggunakan akun ini untuk melakukan 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. Biarkan opsi lain tetap pada setelan defaultnya.

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 project Anda. Misalnya, example-project.

Mengupload DAG

Upload DAG ke lingkungan Anda:

1. Simpan file DAG berikut di komputer lokal Anda.
2. Ganti <PROJECT_ID> dengan Project ID 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, mengambil pesan Pub/Sub, dan memicu DAG lain yang ditentukan dalam ID DAG data pesan Pub/Sub. Dalam contoh ini, trigger_dag memicu DAG target_dag, yang menghasilkan pesan ke log tugas.

DAG trigger_dag berisi 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.

Catatan: Jika ingin meneruskan lebih banyak parameter ke operasi DAG, seperti conf atau execution_date, Anda dapat mengubah fungsi Cloud Run dan contoh DAG. Teruskan nilai dalam pesan Pub/Sub, lalu tangani dalam tugas trigger_target_dag. Untuk mengetahui informasi selengkapnya tentang parameter yang dapat Anda teruskan, lihat dokumentasi untuk TriggerDagRunOperator.

Men-deploy fungsi Cloud Run yang memublikasikan pesan di topik Pub/Sub

Di bagian ini, Anda akan men-deploy fungsi Cloud Run yang memublikasikan pesan di topik Pub/Sub.

Membuat fungsi Cloud Run dan menentukan konfigurasinya

Konsol

1. Di konsol Google Cloud, buka halaman Cloud Run functions.

   Buka fungsi Cloud Run

2. Klik Create function.

3. Di kolom Environment, pilih 1st gen.

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

5. Pada kolom Jenis pemicu, pilih HTTP.

6. Di bagian Autentikasi, pilih Izinkan pemanggilan yang tidak diautentikasi. Opsi ini memberikan kemampuan kepada pengguna yang tidak diautentikasi untuk memanggil fungsi HTTP.

   Perhatian: Untuk penggunaan produksi, pertimbangkan untuk menggunakan fungsi yang memerlukan autentikasi. Lihat Mengautentikasi untuk pemanggilan untuk informasi selengkapnya.

7. Klik Simpan.

8. Klik Berikutnya untuk berpindah ke langkah Kode.

Terraform

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

Contoh ini menunjukkan cara mengupload fungsi Cloud Run dari file arsip zip lokal dengan membuat bucket Cloud Storage, menyimpan file di bucket ini, lalu menggunakan file dari bucket sebagai sumber untuk fungsi Cloud Run. Jika Anda menggunakan pendekatan ini, Terraform tidak akan otomatis memperbarui kode sumber fungsi Anda, 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. Dalam file pubsub_publisher.py, ganti <PROJECT_ID> dengan Project ID project Anda. Misalnya, example-project.
3. Buat arsip zip bernama pubsub_function.zip dengan file pbusub_publisner.py dan 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 fungsi Cloud Run

Konsol

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

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

Terraform

Lewati langkah ini. Parameter fungsi Cloud Run sudah ditentukan dalam resource google_cloudfunctions_function.

Mengupload kode fungsi Cloud Run

Konsol

Di kolom Source code, pilih opsi yang sesuai untuk cara Anda menyediakan kode sumber fungsi. Dalam tutorial ini, tambahkan kode fungsi Anda menggunakan Editor Inline fungsi Cloud Run. 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 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 fungsi Cloud Run sudah ditentukan dalam resource google_cloudfunctions_function.

Menentukan dependensi fungsi Cloud Run

Konsol

Tentukan dependensi fungsi dalam file metadata requirements.txt:

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

Saat Anda men-deploy fungsi, fungsi Cloud Run 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 fungsi Cloud Run ditentukan dalam file requirements.txt dalam arsip pubsub_function.zip.

Men-deploy fungsi Cloud Run

Konsol

Klik Deploy. Setelah deployment berhasil, fungsi akan muncul dengan tanda centang hijau di halaman fungsi Cloud Run di konsol Google Cloud.

Pastikan akun layanan yang menjalankan fungsi Cloud Run Anda memiliki izin yang memadai di project Anda untuk mengakses Pub/Sub.

Terraform

1. Lakukan inisialisasi Terraform:

   terraform init
   

2. Tinjau konfigurasi dan pastikan resource yang akan dibuat atau diupdate 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 yes pada prompt:

   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 fungsi Cloud Run

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

1. Pastikan DAG aktif:

   1. Di konsol Google Cloud, buka halaman Environments.

      Buka Lingkungan

   2. Di daftar lingkungan, klik nama lingkungan Anda. Halaman Environment details akan terbuka.

   3. Buka tab DAG.

   4. Periksa nilai di kolom Status 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 fungsi Cloud Run

   2. Klik nama fungsi Anda, pubsub-publisher.

   3. Buka tab Pengujian.

   4. Di bagian Configure triggering event, masukkan key-value JSON berikut: {"message": "target_dag"}. Jangan ubah pasangan nilai kunci, karena pesan ini akan memicu DAG pengujian nanti.

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

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

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

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

3. Pastikan target_dag telah dipicu:

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

   2. Di konsol Google Cloud, buka halaman Environments.

      Buka Lingkungan

   3. Di daftar lingkungan, klik nama lingkungan Anda. Halaman Environment details akan terbuka.

   4. Buka tab DAG.

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

      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 dalam log operasi DAG. Jika DAG dipicu, tugas trigger_target akan ditandai sebagai Success.

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

   7. Kembali ke tab DAG dan pastikan kolom Operasi yang berhasil untuk DAG target_dag mencantumkan satu operasi yang berhasil.

Ringkasan

Dalam tutorial ini, Anda telah mempelajari cara menggunakan fungsi Cloud Run untuk memublikasikan pesan di topik Pub/Sub dan men-deploy DAG yang berlangganan topik Pub/Sub, mengambil 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 dan memicu DAG yang berada di luar cakupan tutorial ini. Misalnya, Anda dapat menggunakan fungsi Cloud Run untuk memicu DAG Airflow saat peristiwa tertentu terjadi. Lihat tutorial kami untuk mencoba fitur Google Cloud lainnya.

Pembersihan

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

Menghapus project

Perhatian: Menghapus project memiliki efek berikut:

• Semua hal dalam project akan dihapus. Jika menggunakan project yang sudah ada untuk tugas dalam dokumen ini, saat Anda menghapusnya, pekerjaan lain yang telah Anda lakukan dalam project tersebut juga akan terhapus.
• Project ID kustom hilang. Saat membuat project ini, Anda mungkin telah membuat project ID kustom yang ingin digunakan di masa mendatang. Untuk mempertahankan URL yang menggunakan project ID, seperti URL appspot.com, hapus resource yang dipilih di dalam project, bukan menghapus seluruh project.

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

Delete a Google Cloud project:

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. Hapus lingkungan Cloud Composer. Anda juga akan menghapus bucket lingkungan selama prosedur ini.
2. Hapus topik Pub/Sub, dag-topic-trigger.

3. Hapus fungsi Cloud Run.

   1. Di konsol Google Cloud, buka fungsi Cloud Run.

      Buka fungsi Cloud Run

   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 masih ditetapkan (jika Anda menambahkan definisi tersebut ke skrip Terraform).
2. Jalankan terraform destroy.
3. Hapus bucket lingkungan secara manual. Cloud Composer tidak menghapusnya secara otomatis. Anda dapat melakukannya dari konsol Google Cloud atau Google Cloud CLI.

Langkah selanjutnya

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