Menguji, menyinkronkan, dan men-deploy DAG dari GitHub

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Catatan: Halaman ini belum direvisi untuk Cloud Composer 3 dan menampilkan untuk Cloud Composer 2.

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

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

Ringkasan pipeline CI/CD

Gambar 1. Diagram arsitektur yang menampilkan langkah-langkah alurnya (klik untuk memperbesar)

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

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

2. Anda membuka permintaan pull terhadap cabang utama repositori.

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

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

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

6. Anda memverifikasi bahwa DAG berperilaku seperti yang diharapkan dalam pengembangan Anda lingkungan fleksibel App Engine.

7. Jika DAG berfungsi seperti yang diharapkan, Anda harus mengupload DAG ke produksi lingkungan Cloud Composer.

Tujuan

Tujuan:

• Jalankan pemeriksaan pra-pengiriman otomatis menggunakan Cloud Build. Pemeriksaan ini menjalankan pengujian unit untuk DAG.
• Menyinkronkan DAG di lingkungan Cloud Composer pengembangan Anda dengan DAG di repositori GitHub Anda.

Sebelum memulai

• Panduan ini mengasumsikan bahwa Anda bekerja dengan dua model Lingkungan Cloud Composer: lingkungan pengembangan dan lingkungan production.

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

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

  Tujuan contoh pipeline CI/CD menunjukkan isi repositori contoh. DAG dan pengujian disimpan di direktori dags/, dengan file persyaratan, batasan dan file konfigurasi Cloud Build yang disimpan di level teratas. Utilitas sinkronisasi DAG dan persyaratannya terdapat di Direktori utils.

Membuat tugas pemeriksaan pra-pengiriman dan pengujian unit

Tugas Cloud Build pertama menjalankan pemeriksaan pra-pengiriman, yang menjalankan unit untuk DAG.

Menambahkan pengujian unit

Jika Anda belum melakukannya, tulis pengujian unit untuk DAG Anda. Simpan pengujian ini bersama DAG di repositori Anda, masing-masing dengan akhiran _test. Misalnya, jika file untuk DAG di example_dag.py adalah example_dag_test.py. Berikut adalah pengujian yang dijalankan sebagai pemeriksaan pra-pengiriman di repositori Anda.

Membuat konfigurasi YAML Cloud Build untuk pemeriksaan prapengiriman

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

1. Instal dependensi yang diperlukan oleh DAG Anda.
2. Instal dependensi yang diperlukan oleh pengujian unit Anda.
3. Jalankan 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 tentukan tag di kolom name di setiap langkah. Jika Anda menggunakan python sebagai entrypoint, pastikan entrypoint cocok dengan versi yang ditentukan dalam tag Anda, jika tidak, entrypoint akan ditetapkan secara default ke latest.

Membuat pemicu Cloud Build untuk pemeriksaan prapengiriman

Ikuti Membangun repositori dari GitHub panduan 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 menjadi nama cabang dasar repositori, jika diperlukan)

• Sumber - Kontrol Komentar: tidak wajib

  Perhatian: Menyetel Kontrol Komentar tidak diperlukan berarti setiap pengguna dapat mengirimkan Permintaan Pull yang memicu build. Untuk repositori yang lebih besar atau repositori dengan kontributor yang tidak tepercaya, sehingga mewajibkan Kontrol Komentar direkomendasikan.

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

Membuat tugas sinkronisasi DAG dan menambahkan skrip utilitas DAG

Selanjutnya, konfigurasikan tugas Cloud Build yang menjalankan skrip utilitas DAG. Tujuan {i>script<i} utilitas dalam tugas ini menyinkronkan DAG Anda dengan Lingkungan Cloud Composer setelah digabungkan ke cabang utama dalam repositori Anda.

Catatan: Cloud Composer menguraikan semua DAG pada interval yang berulang. Ini terjadi meskipun DAG tidak berubah sejak penguraian terakhir. Karena ini perilaku ini, utilitas ini mengupload semua DAG ke Cloud Composer, bukan hanya mereka 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 ke direktori sementara, dengan mengabaikan semua file Python non-DAG. Tujuan skrip, lalu menggunakan library klien Cloud Storage untuk mengupload semua file dari direktori sementara itu ke direktori dags/ di Bucket lingkungan Cloud Composer.

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 gsutil -m cp -r 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 Cloud Build Anda untuk menyinkronkan DAG. Di dalamnya, ada dua langkah:

1. Instal dependensi yang diperlukan oleh skrip utilitas DAG.

2. Jalankan skrip utilitas untuk menyinkronkan DAG di repositori Anda dengan lingkungan Cloud Composer.

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 tentukan tag di kolom name di setiap langkah. Jika Anda menggunakan python sebagai entrypoint, pastikan entrypoint cocok dengan versi yang ditentukan dalam tag Anda, jika tidak, entrypoint akan ditetapkan secara default ke latest.

Membuat pemicu Cloud Build

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

• Nama: add-dags-to-composer

• Peristiwa: Mengirim ke cabang

• Sumber - Repositori: pilih repositori Anda

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

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

• Konfigurasi Build - File konfigurasi build cloud: /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, kodenya adalah dags/.

• _DAGS_BUCKET - bucket Cloud Storage yang berisi Direktori dags/ di Cloud Composer pengembangan Anda lingkungan fleksibel App Engine. Hapus awalan gs://. Contoh: us-central1-example-env-1234ab56-bucket.

Perhatian: Jika Anda tidak menghapus akhiran dags/ dari Cloud Storage nama bucket dalam variabel substitusi _DAGS_BUCKET, tugas sinkronisasi DAG 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 membuat pemicu Cloud Build.

Menjalankan tugas pra-pengiriman

Buat permintaan pull ke cabang utama untuk menguji build Anda. Temukan pemeriksaan pra-pengiriman di halaman. Klik Details dan pilih Lihat detail selengkapnya tentang Google Cloud Build untuk melihat log build Anda di Konsol Google Cloud Anda.

Penting: Cloud Build menjalankan pengujian unit dengan paket yang ditentukan dalam requirements.txt dan requirements-test.txt yang menyertakan versi publik dari Airflow. Lingkungan Cloud Composer Anda mungkin memiliki sekumpulan paket PyPI dan paket bawaan. Sebaiknya Anda menjalankan lingkungan Airflow lokal sambil mengembangkan DAG.

Gambar 2. Screenshot status pemeriksaan pra-pengiriman Cloud Build di GitHub (klik untuk memperbesar)

Jika pemeriksaan pra-pengiriman Anda gagal, lihat Mengatasi kegagalan build.

Memvalidasi bahwa DAG berfungsi di lingkungan Cloud Composer pengembangan

Setelah permintaan pull disetujui, gabungkan permintaan tersebut ke cabang utama Anda. Gunakan Konsol Google Cloud 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 Cloud Composer pengembangan Anda. Di sana, Anda dapat memvalidasi bahwa DAG berfungsi seperti yang diharapkan.

Menambahkan DAG ke lingkungan produksi

Setelah DAG berperforma seperti yang diharapkan, tambahkan secara manual ke produksi lingkungan fleksibel App Engine. Untuk melakukannya, mengupload file DAG ke direktori dags/ di Cloud Composer produksi Anda bucket lingkungan.

Jika tugas sinkronisasi DAG gagal atau jika DAG tidak berperilaku seperti yang diharapkan dalam lingkungan Cloud Composer pengembangan, lihat Mengatasi kegagalan build.

Mengatasi kegagalan build

Bagian ini menjelaskan cara mengatasi skenario kegagalan build yang umum.

Bagaimana jika pemeriksaan pra-pengiriman saya gagal?

Dari permintaan pull, klik Details lalu pilih Lihat detail selengkapnya tentang Google Cloud Build untuk melihat log build Anda di Konsol Google Cloud Anda. Gunakan log ini untuk membantu Anda men-debug masalah dengan DAG. Setelah masalah teratasi, lakukan perbaikan dan kirim ke . Pemeriksaan pra-pengiriman akan berjalan kembali, dan Anda dapat terus melakukan iterasi menggunakan log sebagai alat debug.

Bagaimana jika tugas sinkronisasi DAG saya gagal?

Gunakan Konsol Google Cloud 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 selesaikan masalah yang sama. Jika Anda memerlukan bantuan tambahan untuk mengatasi error ini, saluran dukungan.

Bagaimana jika DAG saya tidak berfungsi dengan baik di lingkungan Cloud Composer?

Jika DAG tidak berfungsi seperti yang diharapkan dalam pengembangan Anda di lingkungan Cloud Composer, jangan promosikan DAG secara manual ke lingkungan Cloud Composer production. Sebagai gantinya, lakukan salah satu hal berikut:

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

Mengikuti salah satu langkah ini akan memicu pemeriksaan pra-pengiriman baru dan saat penggabungan, tugas sinkronisasi DAG.

Langkah selanjutnya

• Menjalankan lingkungan Airflow lokal
• Menulis DAG
• Memicu DAG
• DAG Pengujian