Menguji, menyinkronkan, dan men-deploy DAG dari GitHub

Managed Airflow (Gen 3) | Managed Airflow (Gen 2) | Managed Airflow (Legacy Gen 1)

Panduan ini menjelaskan cara membuat pipeline CI/CD untuk menguji, menyinkronkan, dan men-deploy DAG ke lingkungan Managed Airflow Anda dari repositori GitHub.

Jika Anda hanya ingin menyinkronkan data dari layanan lain, lihat Mentransfer data dari layanan lain.

Ringkasan pipeline CI/CD

Gambar 1. Diagram arsitektur yang menunjukkan langkah-langkah alur (klik untuk memperbesar)

Pipeline CI/CD untuk menguji, menyinkronkan, dan men-deploy DAG memiliki langkah-langkah berikut:

1. Anda membuat perubahan pada DAG dan mengirimkan perubahan tersebut ke cabang pengembangan di repositori Anda.

2. Anda membuka permintaan pull terhadap cabang utama repositori Anda.

3. Cloud Build menjalankan pengujian unit untuk memeriksa apakah DAG Anda valid.

4. Permintaan pull Anda disetujui dan digabungkan ke cabang utama repositori Anda.

5. Cloud Build menyinkronkan lingkungan Managed Airflow pengembangan Anda dengan perubahan baru ini.

6. Anda memverifikasi bahwa DAG berfungsi seperti yang diharapkan di lingkungan pengembangan Anda.

7. Jika DAG Anda berfungsi seperti yang diharapkan, upload DAG ke lingkungan Managed Airflow produksi Anda.

Tujuan

• Menjalankan pemeriksaan presubmit otomatis menggunakan Cloud Build. Pemeriksaan ini menjalankan pengujian unit untuk DAG.
• Menyinkronkan DAG di lingkungan Managed Service untuk Apache Airflow pengembangan Anda dengan DAG di repositori GitHub Anda.

Sebelum memulai

• Panduan ini mengasumsikan bahwa Anda menggunakan dua lingkungan Managed Airflow yang identik: lingkungan pengembangan dan lingkungan produksi.

  Untuk tujuan panduan ini, Anda hanya mengonfigurasi pipeline CI/CD untuk lingkungan pengembangan. Pastikan lingkungan yang Anda gunakan bukan lingkungan produksi.

• Panduan ini mengasumsikan bahwa Anda telah menyimpan DAG dan pengujiannya di repositori GitHub.

  Pipeline CI/CD contoh menunjukkan konten repositori contoh. DAG dan pengujian disimpan di direktori dags/, dengan file persyaratan, file batasan, dan file konfigurasi Cloud Build yang disimpan di tingkat atas. Utilitas sinkronisasi DAG dan persyaratannya terletak di direktori utils.

Membuat tugas pemeriksaan presubmit dan pengujian unit

Tugas Cloud Build pertama menjalankan pemeriksaan presubmit, yang mengeksekusi pengujian unit untuk DAG Anda.

Menambahkan pengujian unit

Jika belum, buat pengujian unit untuk DAG Anda. Simpan pengujian ini bersama DAG di repositori Anda, masing-masing dengan akhiran _test. Misalnya, file pengujian untuk DAG di example_dag.py adalah example_dag_test.py. Ini adalah pengujian yang berjalan sebagai pemeriksaan presubmit di repositori Anda.

Membuat konfigurasi YAML Cloud Build untuk pemeriksaan presubmit

Di repositori Anda, buat file YAML bernama test-dags.cloudbuild.yaml yang mengonfigurasi tugas Cloud Build Anda untuk pemeriksaan presubmit. Di dalamnya, ada tiga langkah:

1. Menginstal dependensi yang diperlukan oleh DAG Anda.
2. Menginstal dependensi yang diperlukan oleh pengujian unit Anda.
3. Mengeksekusi pengujian DAG.

steps:
  # install dependencies
  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements.txt", "-c", "constraints.txt", "--user"]

  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements-test.txt", "--user"]

  # run in python 3.8 which is latest version in Cloud Composer
  - name: python:3.8-slim
    entrypoint: python3.8
    args: ["-m", "pytest", "-s", "dags/"]

Catatan: Cloud Build mengambil image Python dari Docker Hub dan Anda dapat menentukan tag di kolom name setiap langkah. Jika Anda menggunakan python sebagai entrypoint, pastikan entrypoint Anda cocok dengan versi yang ditentukan dalam tag Anda, jika tidak, entrypoint akan ditetapkan secara default ke latest.

Membuat pemicu Cloud Build untuk pemeriksaan presubmit

Ikuti panduan Membangun repositori dari GitHub untuk membuat pemicu berbasis aplikasi GitHub dengan konfigurasi berikut:

• Nama: test-dags

• Peristiwa: Permintaan Pull

• Sumber - Repositori: pilih repositori Anda

• Sumber - Cabang dasar: ^main$ (ubah main ke nama cabang dasar repositori Anda, jika diperlukan)

• Sumber - Kontrol Komentar: tidak diperlukan

  Perhatian: Menetapkan Kontrol Komentar agar tidak diperlukan berarti pengguna mana pun dapat mengirimkan Permintaan Pull yang memicu build. Untuk repositori yang lebih besar atau repositori dengan kontributor yang tidak tepercaya, sebaiknya Kontrol Komentar diperlukan.

• Konfigurasi Build - File konfigurasi Cloud Build: /test-dags.cloudbuild.yaml (jalur ke file build Anda)

Membuat tugas sinkronisasi DAG dan menambahkan skrip utilitas DAG

Selanjutnya, konfigurasi tugas Cloud Build yang menjalankan skrip utilitas DAG. Skrip utilitas dalam tugas ini menyinkronkan DAG Anda dengan lingkungan Managed Airflow Anda setelah digabungkan ke cabang utama di repositori Anda.

Catatan: Managed Airflow mengurai semua DAG pada interval berulang. Hal ini terjadi meskipun DAG tidak berubah sejak penguraian terakhir. Karena perilaku ini, utilitas ini mengupload semua DAG ke Managed Airflow, bukan hanya DAG yang telah berubah.

Menambahkan skrip utilitas DAG

Tambahkan skrip utilitas DAG ke repositori Anda. Skrip utilitas ini menyalin semua file DAG di direktori dags/ repositori Anda ke direktori sementara, dengan mengabaikan semua file Python non-DAG. Kemudian, skrip menggunakan library klien Cloud Storage untuk mengupload semua file dari direktori sementara tersebut ke direktori dags/ di bucket lingkungan Managed Airflow Anda.

from __future__ import annotations

import argparse
import glob
import os
from shutil import copytree, ignore_patterns
import tempfile

# Imports the Google Cloud client library
from google.cloud import storage


def _create_dags_list(dags_directory: str) -> tuple[str, list[str]]:
    temp_dir = tempfile.mkdtemp()

    # ignore non-DAG Python files
    files_to_ignore = ignore_patterns("__init__.py", "*_test.py")

    # Copy everything but the ignored files to a temp directory
    copytree(dags_directory, f"{temp_dir}/", ignore=files_to_ignore, dirs_exist_ok=True)

    # The only Python files left in our temp directory are DAG files
    # so we can exclude all non Python files
    dags = glob.glob(f"{temp_dir}/*.py")
    return (temp_dir, dags)


def upload_dags_to_composer(
    dags_directory: str, bucket_name: str, name_replacement: str = "dags/"
) -> None:
    """
    Given a directory, this function moves all DAG files from that directory
    to a temporary directory, then uploads all contents of the temporary directory
    to a given cloud storage bucket
    Args:
        dags_directory (str): a fully qualified path to a directory that contains a "dags/" subdirectory
        bucket_name (str): the GCS bucket of the Cloud Composer environment to upload DAGs to
        name_replacement (str, optional): the name of the "dags/" subdirectory that will be used when constructing the temporary directory path name Defaults to "dags/".
    """
    temp_dir, dags = _create_dags_list(dags_directory)

    if len(dags) > 0:
        # Note - the GCS client library does not currently support batch requests on uploads
        # if you have a large number of files, consider using
        # the Python subprocess module to run gcloud storage cp --recursive on your dags
        # See https://cloud.google.com/storage/docs/gsutil/commands/cp for more info
        storage_client = storage.Client()
        bucket = storage_client.bucket(bucket_name)

        for dag in dags:
            # Remove path to temp dir
            dag = dag.replace(f"{temp_dir}/", name_replacement)

            try:
                # Upload to your bucket
                blob = bucket.blob(dag)
                blob.upload_from_filename(dag)
                print(f"File {dag} uploaded to {bucket_name}/{dag}.")
            except FileNotFoundError:
                current_directory = os.listdir()
                print(
                    f"{name_replacement} directory not found in {current_directory}, you may need to override the default value of name_replacement to point to a relative directory"
                )
                raise

    else:
        print("No DAGs to upload.")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--dags_directory",
        help="Relative path to the source directory containing your DAGs",
    )
    parser.add_argument(
        "--dags_bucket",
        help="Name of the DAGs bucket of your Composer environment without the gs:// prefix",
    )

    args = parser.parse_args()

    upload_dags_to_composer(args.dags_directory, args.dags_bucket)

Membuat konfigurasi YAML Cloud Build untuk menyinkronkan DAG

Di repositori Anda, buat file YAML bernama add-dags-to-composer.cloudbuild.yaml yang mengonfigurasi tugas Cloud Build Anda untuk menyinkronkan DAG. Di dalamnya, ada dua langkah:

1. Menginstal dependensi yang diperlukan oleh skrip utilitas DAG.

2. Menjalankan skrip utilitas untuk menyinkronkan DAG di repositori Anda dengan lingkungan Managed Airflow Anda.

steps:
  # install dependencies
  - name: python
    entrypoint: pip
    args: ["install", "-r", "utils/requirements.txt", "--user"]

  # run
  - name: python
    entrypoint: python
    args: ["utils/add_dags_to_composer.py", "--dags_directory=${_DAGS_DIRECTORY}", "--dags_bucket=${_DAGS_BUCKET}"]

Catatan: Cloud Build mengambil image Python dari Docker Hub dan Anda dapat menentukan tag di kolom name setiap langkah. Jika Anda menggunakan python sebagai entrypoint, pastikan entrypoint Anda cocok dengan versi yang ditentukan dalam tag Anda, jika tidak, entrypoint akan ditetapkan secara default ke latest.

Membuat pemicu Cloud Build

Ikuti panduan Membangun repositori dari GitHub untuk membuat pemicu berbasis aplikasi GitHub dengan konfigurasi berikut:

• Nama: add-dags-to-composer

• Peristiwa: Kirim ke cabang

• Sumber - Repositori: pilih repositori Anda

• Sumber - Cabang dasar: ^main$ (ubah main ke nama cabang dasar repositori Anda, jika diperlukan)

• Sumber - Filter file yang disertakan (glob): dags/**

• Konfigurasi Build - File konfigurasi Cloud Build: /add-dags-to-composer.cloudbuild.yaml (jalur ke file build Anda)

Di Konfigurasi lanjutan, tambahkan dua variabel substitusi:

• _DAGS_DIRECTORY - direktori tempat DAG berada di repositori Anda. Jika Anda menggunakan repositori contoh dari panduan ini, direktori tersebut adalah dags/.

• _DAGS_BUCKET - bucket Cloud Storage yang berisi direktori dags/ di lingkungan Managed Airflow pengembangan Anda. Hapus awalan gs://. Contoh: us-central1-example-env-1234ab56-bucket.

Perhatian: Jika Anda tidak menghapus akhiran dags/ dari nama bucket Cloud Storage di variabel substitusi _DAGS_BUCKET, tugas sinkronisasi DAG akan gagal dengan error NotFound. Hal ini terjadi karena di Cloud Storage, dags/ adalah folder, yang dianggap sebagai bagian dari nama objek, bukan bagian dari nama bucket.

Menguji pipeline CI/CD

Di bagian ini, ikuti alur pengembangan DAG yang menggunakan pemicu Cloud Build yang baru Anda buat.

Menjalankan tugas presubmit

Buat permintaan pull ke cabang utama Anda untuk menguji build Anda. Temukan pemeriksaan presubmit Anda di halaman. Klik Details , lalu pilih View more details on Google Cloud Build untuk melihat log build Anda di Google Cloud konsol.

Penting: Cloud Build menjalankan pengujian unit dengan paket yang ditentukan dalam requirements.txt dan requirements-test.txt, yang mencakup versi publik Airflow. Lingkungan Managed Airflow Anda mungkin memiliki kumpulan paket PyPI dan paket yang telah diinstal sebelumnya yang berbeda. Sebaiknya jalankan lingkungan Airflow lokal saat mengembangkan DAG Anda.

Gambar 2. Screenshot status pemeriksaan presubmit Cloud Build di GitHub (klik untuk memperbesar)

Jika pemeriksaan presubmit Anda gagal, lihat Mengatasi kegagalan build.

Memvalidasi bahwa DAG Anda berfungsi di lingkungan pengembangan Anda

Setelah permintaan pull Anda disetujui, gabungkan ke cabang utama Anda. Gunakan Google Cloud konsol untuk melihat hasil build Anda. Jika Anda memiliki banyak pemicu Cloud Build, Anda dapat memfilter build berdasarkan nama pemicu add-dags-to-composer.

Setelah tugas sinkronisasi Cloud Build berhasil, DAG yang disinkronkan akan muncul di lingkungan Managed Airflow pengembangan Anda. Di sana, Anda dapat memvalidasi bahwa fungsi DAG berfungsi seperti yang diharapkan.

Menambahkan DAG ke lingkungan produksi Anda

Setelah DAG berfungsi seperti yang diharapkan, tambahkan secara manual ke lingkungan produksi Anda. Untuk melakukannya, upload file DAG ke direktori dags/ di lingkungan Managed Airflow produksi Anda bucket.

Jika tugas sinkronisasi DAG Anda gagal atau jika DAG Anda tidak berfungsi seperti yang diharapkan di lingkungan Managed Airflow pengembangan Anda, lihat Mengatasi kegagalan build.

Mengatasi kegagalan build

Bagian ini menjelaskan cara mengatasi skenario kegagalan build umum.

Bagaimana jika pemeriksaan presubmit saya gagal?

Dari permintaan pull Anda, klik Details , lalu pilih View more details on Google Cloud Build untuk melihat log build Anda di Google Cloud konsol. Gunakan log ini untuk membantu Anda men-debug masalah pada DAG Anda. Setelah Anda mengatasi masalah, lakukan commit perbaikan dan kirim ke cabang Anda. Pemeriksaan presubmit akan berjalan lagi, dan Anda dapat terus melakukan iterasi menggunakan log sebagai alat proses debug.

Bagaimana jika tugas sinkronisasi DAG saya gagal?

Gunakan Google Cloud konsol untuk melihat hasil build Anda. Jika Anda memiliki banyak pemicu Cloud Build, Anda dapat memfilter build berdasarkan nama pemicu add-dags-to-composer. Periksa log tugas build dan atasi error. Jika Anda memerlukan bantuan tambahan untuk mengatasi error, gunakan saluran dukungan.

Bagaimana jika DAG saya tidak berfungsi dengan benar di lingkungan Managed Airflow saya?

Jika DAG Anda tidak berfungsi seperti yang diharapkan di lingkungan Managed Airflow pengembangan Anda, jangan promosikan DAG secara manual ke lingkungan Managed Airflow produksi Anda. Sebagai gantinya, lakukan salah satu hal berikut:

• Kembalikan permintaan pull dengan perubahan yang merusak DAG Anda untuk memulihkannya ke status segera sebelum perubahan Anda (tindakan ini juga mengembalikan semua file lain dalam permintaan pull tersebut).
• Buat permintaan pull baru untuk mengembalikan perubahan secara manual ke DAG yang rusak.
• Buat permintaan pull baru untuk memperbaiki error di DAG Anda.

Mengikuti salah satu langkah ini akan memicu pemeriksaan presubmit baru dan setelah penggabungan, tugas sinkronisasi DAG.

Langkah berikutnya

• Menjalankan lingkungan Airflow lokal
• Menulis DAG
• Menjadwalkan dan memicu DAG
• Menguji DAG