Menambahkan dan memperbarui DAG

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Halaman ini menjelaskan cara mengelola DAG di lingkungan Cloud Composer Anda.

Cloud Composer menggunakan bucket Cloud Storage untuk menyimpan DAG lingkungan Cloud Composer Anda. Lingkungan Anda menyinkronkan DAG dari bucket ini ke komponen Airflow seperti worker dan penjadwal Airflow.

Sebelum memulai

• Karena Apache Airflow tidak menyediakan isolasi DAG yang kuat, sebaiknya Anda mempertahankan lingkungan produksi dan pengujian yang terpisah untuk mencegah gangguan DAG. Untuk mengetahui informasi selengkapnya, lihat Menguji DAG.
• Pastikan akun Anda memiliki izin yang cukup untuk mengelola DAG.
• Perubahan pada DAG diterapkan ke Airflow dalam waktu 3-5 menit. Anda dapat melihat status tugas di antarmuka web Airflow.

Mengakses bucket lingkungan Anda

Untuk mengakses bucket yang terkait dengan lingkungan Anda:

gcloud composer environments describe ENVIRONMENT_NAME \
    --location LOCATION \
    --format="get(config.dagGcsPrefix)"

gcloud beta composer environments describe example-environment \
    --location us-central1 \
    --format="get(config.dagGcsPrefix)"

GET https://composer.googleapis.com/v1/projects/example-project/
locations/us-central1/environments/example-environment

import google.auth
import google.auth.transport.requests

# Authenticate with Google Cloud.
# See: https://cloud.google.com/docs/authentication/getting-started
credentials, _ = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
authed_session = google.auth.transport.requests.AuthorizedSession(credentials)

# project_id = 'YOUR_PROJECT_ID'
# location = 'us-central1'
# composer_environment = 'YOUR_COMPOSER_ENVIRONMENT_NAME'

environment_url = (
    "https://composer.googleapis.com/v1beta1/projects/{}/locations/{}"
    "/environments/{}"
).format(project_id, location, composer_environment)
response = authed_session.request("GET", environment_url)
environment_data = response.json()

# Print the bucket name from the response body.
print(environment_data["config"]["dagGcsPrefix"])

Menambahkan atau memperbarui DAG

Untuk menambahkan atau memperbarui DAG, pindahkan file .py Python untuk DAG ke folder /dags di bucket lingkungan.

gcloud composer environments storage dags import \
    --environment ENVIRONMENT_NAME \
    --location LOCATION \
    --source="LOCAL_FILE_TO_UPLOAD"

gcloud composer environments storage dags import \
    --environment example-environment \
    --location us-central1 \
    --source="example_dag.py"

Memperbarui DAG yang memiliki operasi DAG aktif

Jika Anda mengupdate DAG yang memiliki operasi DAG aktif:

• Semua tugas yang sedang berjalan akan selesai menggunakan file DAG asli.
• Semua tugas yang dijadwalkan tetapi saat ini tidak berjalan menggunakan file DAG yang telah diupdate.
• Semua tugas yang tidak lagi ada dalam file DAG yang diperbarui ditandai sebagai dihapus.

Memperbarui DAG yang berjalan sesuai jadwal rutin

Setelah Anda mengupload file DAG, Airflow memerlukan waktu beberapa saat untuk memuat file ini dan memperbarui DAG. Jika DAG berjalan sesuai jadwal yang sering, Anda mungkin ingin memastikan bahwa DAG menggunakan versi file DAG yang telah diupdate. Untuk melakukannya:

1. Jeda DAG di UI Airflow.

2. Upload file DAG yang diperbarui.

3. Tunggu hingga Anda melihat update di UI Airflow. Artinya, DAG diurai dengan benar oleh scheduler dan diperbarui di database Airflow.

   Jika UI Airflow menampilkan DAG yang telah diupdate, hal ini tidak menjamin bahwa worker Airflow memiliki versi file DAG yang telah diupdate. Hal ini terjadi karena file DAG disinkronkan secara independen untuk penjadwal dan pekerja.

4. Anda mungkin ingin memperpanjang waktu tunggu untuk memastikan file DAG disinkronkan dengan semua pekerja di lingkungan Anda. Sinkronisasi terjadi beberapa kali per menit. Dalam lingkungan yang sehat, menunggu sekitar 20-30 detik sudah cukup bagi semua pekerja untuk melakukan sinkronisasi.

5. (Opsional) Jika Anda ingin memastikan sepenuhnya bahwa semua pekerja memiliki file DAG versi baru, periksa log untuk setiap pekerja. Untuk melakukannya:

   1. Buka tab Log untuk lingkungan Anda di konsol Google Cloud .

   2. Buka Composer logs > Infrastructure > Cloud Storage sync, lalu periksa log untuk setiap pekerja di lingkungan Anda. Cari item log Syncing dags directory terbaru yang memiliki stempel waktu setelah Anda mengupload file DAG baru. Jika Anda melihat item Finished syncing yang mengikutinya, berarti DAG berhasil disinkronkan di worker ini.

6. Lanjutkan DAG.

Mengurai ulang DAG

Karena DAG disimpan di bucket lingkungan, setiap DAG disinkronkan ke pemroses DAG terlebih dahulu, lalu pemroses DAG akan memparsingnya dengan sedikit penundaan. Jika Anda mem-parsing ulang DAG secara manual, misalnya, melalui UI Airflow, maka pemroses DAG mem-parsing ulang versi DAG saat ini yang tersedia untuknya, yang mungkin bukan versi DAG terbaru yang Anda upload ke bucket lingkungan.

Sebaiknya gunakan penguraian ulang sesuai permintaan hanya jika Anda mengalami waktu penguraian yang lama. Misalnya, hal ini dapat terjadi jika lingkungan Anda memiliki banyak file, atau jika interval parsing DAG yang panjang dikonfigurasi dalam opsi konfigurasi Airflow.

Menghapus DAG di lingkungan Anda

Untuk menghapus DAG, hapus file .py Python untuk DAG dari folder /dags lingkungan di bucket lingkungan Anda.

gcloud composer environments storage dags delete \
    --environment ENVIRONMENT_NAME \
    --location LOCATION \
    DAG_FILE

gcloud composer environments storage dags delete \
    --environment example-environment \
    --location us-central1 \
    example_dag.py

Menghapus DAG dari UI Airflow

Untuk menghapus metadata DAG dari antarmuka web Airflow:

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags delete -- DAG_NAME

Langkah berikutnya

• Menulis DAG
• Menguji, menyinkronkan, dan men-deploy DAG dari GitHub
• Memecahkan masalah DAG