Menggunakan operator yang dapat ditangguhkan di DAG

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Halaman ini menjelaskan cara mengaktifkan dukungan untuk Operator yang Dapat Ditunda di lingkungan Anda dan menggunakan operator Google Cloud yang dapat ditangguhkan di DAG Anda.

Tentang Operator yang Dapat Ditunda di Cloud Composer

Jika memiliki minimal satu instance pemicu (atau minimal dua di lingkungan yang sangat tangguh), Anda dapat menggunakan Operator dan Pemicu yang Dapat Ditunda di DAG.

Untuk operator yang dapat ditangguhkan, Airflow membagi eksekusi tugas menjadi tahap berikut:

  1. Mulai operasi. Pada tahap ini, tugas menempati slot pekerja Airflow. Tugas ini melakukan operasi yang mendelegasikan tugas ke layanan lain.

    Misalnya, menjalankan tugas BigQuery dapat memerlukan waktu dari beberapa detik hingga beberapa jam. Setelah membuat tugas, operasi ini akan meneruskan ID tugas (ID tugas BigQuery) ke pemicu Airflow.

  2. Pemicu memantau tugas hingga selesai. Pada tahap ini, slot pekerja tidak terisi. Pemicu Airflow memiliki arsitektur asinkron dan mampu menangani ratusan tugas tersebut. Saat pemicu mendeteksi bahwa tugas selesai, pemicu akan mengirim peristiwa yang memicu tahap terakhir.

  3. Pada tahap terakhir, pekerja Airflow akan mengeksekusi callback. Misalnya, callback ini dapat menandai tugas sebagai berhasil, atau menjalankan operasi lain dan menetapkan tugas untuk dipantau lagi oleh pemicu.

Pemicu bersifat stateless sehingga tahan terhadap gangguan atau mulai ulang. Oleh karena itu, tugas yang berjalan lama tahan terhadap mulai ulang pod, kecuali jika mulai ulang terjadi selama tahap terakhir, yang diperkirakan akan berlangsung singkat.

Sebelum memulai

  • Operator dan Sensor yang Dapat Ditunda tersedia di lingkungan Cloud Composer 2 dan memerlukan hal berikut:
    • Cloud Composer 2.0.31 dan versi yang lebih baru
    • Airflow 2.2.5, 2.3.3, dan versi yang lebih baru

Mengaktifkan dukungan untuk operator yang dapat ditangguhkan

Komponen lingkungan yang disebut pemicu Airflow secara asinkron memantau semua tugas yang ditangguhkan di lingkungan Anda. Setelah operasi yang ditangguhkan dari tugas tersebut selesai, pemicu akan meneruskan tugas ke pekerja Airflow.

Anda memerlukan setidaknya satu instance pemicu di lingkungan (atau setidaknya dua di lingkungan yang sangat tangguh) untuk menggunakan mode yang dapat ditangguhkan di DAG. Anda dapat mengonfigurasi pemicu saat membuat lingkungan atau menyesuaikan jumlah pemicu dan parameter performa untuk lingkungan yang ada.

Operator Google Cloud yang mendukung mode yang dapat ditangguhkan

Hanya beberapa operator Airflow yang telah diperluas untuk mendukung model yang dapat ditangguhkan. Daftar berikut adalah referensi untuk operator dalam paket airflow.providers.google.operators.cloud yang mendukung mode yang dapat ditangguhkan. Kolom dengan versi paket airflow.providers.google.operators.cloud minimum yang diperlukan mewakili versi paket paling awal tempat operator tersebut mendukung mode yang dapat ditangguhkan.

Operator Cloud Composer

Nama operatorVersi apache-airflow-providers-google yang diperlukan
CloudComposerCreateEnvironmentOperator 6.4.0
CloudComposerDeleteEnvironmentOperator 6.4.0
CloudComposerUpdateEnvironmentOperator 6.4.0

Operator BigQuery

Nama operatorVersi apache-airflow-providers-google yang diperlukan
BigQueryCheckOperator 8.4.0
BigQueryValueCheckOperator 8.4.0
BigQueryIntervalCheckOperator 8.4.0
BigQueryGetDataOperator 8.4.0
BigQueryInsertJobOperator 8.4.0

Operator BigQuery Data Transfer Service

Nama operatorVersi apache-airflow-providers-google yang diperlukan
BigQueryDataTransferServiceStartTransferRunsOperator 8.9.0

Operator Cloud Build

Nama operatorVersi apache-airflow-providers-google yang diperlukan
CloudBuildCreateBuildOperator 8.7.0

Operator Cloud SQL

Nama operatorVersi apache-airflow-providers-google yang diperlukan
CloudSQLExportInstanceOperator 10.3.0

Operator Dataflow

Nama operatorVersi apache-airflow-providers-google yang diperlukan
DataflowTemplatedJobStartOperator 8.9.0
DataflowStartFlexTemplateOperator 8.9.0

Operator Cloud Data Fusion

Nama operatorVersi apache-airflow-providers-google yang diperlukan
CloudDataFusionStartPipelineOperator 8.9.0

Operator Google Kubernetes Engine

Nama operatorVersi apache-airflow-providers-google yang diperlukan
GKEDeleteClusterOperator 9.0.0
GKECreateClusterOperator 9.0.0

Operator AI Platform

Nama operatorVersi apache-airflow-providers-google yang diperlukan
MLEngineStartTrainingJobOperator 8.9.0

Menggunakan operator yang dapat ditangguhkan di DAG

Konvensi umum untuk semua operator Google Cloud adalah mengaktifkan mode yang dapat ditangguhkan dengan parameter boolean deferrable. Jika operator Google Cloud tidak memiliki parameter ini, operator tersebut tidak dapat berjalan dalam mode yang dapat ditangguhkan. Operator lain dapat memiliki konvensi yang berbeda. Misalnya, beberapa operator komunitas memiliki class terpisah dengan akhiran Async dalam namanya.

Contoh DAG berikut menggunakan operator DataprocSubmitJobOperator dalam mode yang dapat ditangguhkan:

PYSPARK_JOB = {
    "reference": { "project_id": "PROJECT_ID" },
    "placement": { "cluster_name": "PYSPARK_CLUSTER_NAME" },
    "pyspark_job": {
        "main_python_file_uri": "gs://dataproc-examples/pyspark/hello-world/hello-world.py"
    },
}

DataprocSubmitJobOperator(
        task_id="dataproc-deferrable-example",
        job=PYSPARK_JOB,
        deferrable=True,
    )

Melihat log pemicu

Pemicu menghasilkan log yang tersedia bersama dengan log komponen lingkungan lainnya. Untuk mengetahui informasi selengkapnya tentang cara melihat log lingkungan, lihat Melihat log.

Pemicu monitor

Untuk informasi selengkapnya tentang cara memantau komponen pemicu, lihat Metrik Airflow.

Selain memantau pemicu, Anda dapat memeriksa jumlah tugas yang ditangguhkan dalam metrik Tugas yang Belum Selesai di dasbor Monitoring lingkungan Anda.

Langkah selanjutnya