Menambahkan dan memperbarui DAG

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

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 pekerja dan penjadwal Airflow.

Sebelum memulai

  • Karena Apache Airflow tidak memberikan isolasi DAG yang kuat, sebaiknya Anda mempertahankan lingkungan produksi dan pengujian terpisah untuk mencegah gangguan DAG. Untuk informasi selengkapnya, lihat Menguji DAG.
  • Pastikan akun Anda memiliki izin yang memadai untuk mengelola DAG.
  • Perubahan pada DAG akan 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:

Konsol

  1. Di konsol Google Cloud, buka halaman Environments.

    Buka Lingkungan

  2. Dalam daftar lingkungan, temukan baris dengan nama lingkungan Anda, lalu di kolom DAGs folder, klik link DAGs. Halaman Detail bucket akan terbuka. File ini menampilkan konten folder /dags di bucket lingkungan Anda.

gcloud

gcloud CLI memiliki perintah terpisah untuk menambahkan dan menghapus DAG di bucket lingkungan Anda.

Jika ingin berinteraksi dengan bucket lingkungan, Anda juga dapat menggunakan Google Cloud CLI. Untuk mendapatkan alamat bucket lingkungan Anda, jalankan perintah gcloud CLI berikut:

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

Ganti:

  • ENVIRONMENT_NAME dengan nama lingkungan.
  • LOCATION dengan region tempat lingkungan berada.

Contoh:

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

API

Buat permintaan API environments.get. Di resource Environment, di resource EnvironmentConfig, di resource dagGcsPrefix adalah alamat bucket lingkungan Anda.

Contoh:

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

Python

Gunakan library google-auth untuk mendapatkan kredensial dan gunakan library requests untuk memanggil REST API. 

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.

Konsol

  1. Di konsol Google Cloud, buka halaman Environments.

    Buka Lingkungan

  2. Dalam daftar lingkungan, temukan baris dengan nama lingkungan Anda, lalu di kolom DAGs folder, klik link DAGs. Halaman Detail bucket akan terbuka. File ini menampilkan konten folder /dags di bucket lingkungan Anda.

  3. Klik Upload files. Kemudian, pilih file .py Python untuk DAG menggunakan dialog browser dan konfirmasi.

gcloud 

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

Ganti:

  • ENVIRONMENT_NAME dengan nama lingkungan.
  • LOCATION dengan region tempat lingkungan berada.
  • LOCAL_FILE_TO_UPLOAD adalah file .py Python untuk DAG.

Contoh:

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 dieksekusi akan selesai menggunakan file DAG asli.
  • Semua tugas yang dijadwalkan, tetapi saat ini tidak berjalan, menggunakan file DAG yang diperbarui.
  • Semua tugas yang tidak lagi ada dalam file DAG yang diperbarui akan ditandai sebagai dihapus.

Memperbarui DAG yang berjalan sesuai jadwal yang sering

Setelah Anda mengupload file DAG, perlu waktu beberapa saat bagi Airflow untuk memuat file ini dan memperbarui DAG. Jika DAG berjalan sesuai jadwal yang sering, sebaiknya pastikan DAG menggunakan file DAG versi yang diperbarui. Untuk melakukannya:

  1. Jeda DAG di UI Airflow.

  2. Upload file DAG yang telah diperbarui.

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

    Jika UI Airflow menampilkan DAG yang diperbarui, hal ini tidak menjamin bahwa pekerja Airflow memiliki versi file DAG yang diperbarui. 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 benar-benar yakin bahwa semua pekerja memiliki file DAG versi baru, periksa log untuk setiap pekerja. Untuk melakukannya:

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

    2. Buka item Composer logs > Infrastructure > Cloud Storage sync dan 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 pekerja ini.

  6. Batalkan jeda DAG.

Menghapus DAG di lingkungan Anda

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

Konsol

  1. Di konsol Google Cloud, buka halaman Environments.

    Buka Lingkungan

  2. Dalam daftar lingkungan, temukan baris dengan nama lingkungan Anda, lalu di kolom DAGs folder, klik link DAGs. Halaman Detail bucket akan terbuka. Tindakan ini akan menampilkan konten folder /dags di bucket lingkungan Anda.

  3. Pilih file DAG, klik Hapus, lalu konfirmasi operasinya.

gcloud 

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

Ganti:

  • ENVIRONMENT_NAME dengan nama lingkungan.
  • LOCATION dengan region tempat lingkungan berada.
  • DAG_FILE dengan file .py Python untuk DAG.

Contoh:

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:

UI Airflow

  1. Buka UI Airflow untuk lingkungan Anda.
  2. Untuk DAG, klik Hapus DAG.

gcloud

Di Airflow 1 versi yang lebih lama dari 1.14.0, jalankan perintah berikut di gcloud CLI:

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    delete_dag -- DAG_NAME

Di Airflow 2, Airflow 1.14.0, dan versi yang lebih baru, jalankan perintah berikut di gcloud CLI:

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

Ganti:

  • ENVIRONMENT_NAME dengan nama lingkungan.
  • LOCATION dengan region tempat lingkungan tersebut berada.
  • DAG_NAME adalah nama DAG yang akan dihapus.

Langkah selanjutnya