Cloud Composer 1 | Cloud Composer 2
Panduan ini menunjukkan cara menulis directed acyclic graph (DAG) Apache Airflow yang berjalan di lingkungan Cloud Composer.
Karena Apache Airflow tidak menyediakan DAG dan isolasi tugas yang kuat, sebaiknya gunakan lingkungan produksi dan pengujian yang terpisah untuk mencegah interferensi DAG. Untuk informasi selengkapnya, lihat Menguji DAG.
Menyusun DAG Aliran Udara
DAG Airflow ditentukan dalam file Python dan terdiri dari komponen berikut:
- Definisi DAG
- Operator Airflow
- Hubungan operator
Cuplikan kode berikut menunjukkan contoh setiap komponen di luar konteks.
Definisi DAG
Contoh berikut menunjukkan definisi DAG:
Operator dan tugas
Operator menjelaskan pekerjaan yang akan dilakukan. Tugas adalah instance spesifik dari operator.
Hubungan tugas
Hubungan tugas menjelaskan urutan penyelesaian pekerjaan.
Contoh alur kerja DAG lengkap di Python
Alur kerja berikut adalah template DAG yang berfungsi dan terdiri dari dua tugas: tugas hello_python
dan tugas goodbye_bash
:
Untuk mengetahui informasi selengkapnya tentang cara menentukan DAG Airflow, lihat tutorial Airflow dan Konsep Airflow.
Operator Airflow
Contoh berikut menunjukkan beberapa operator Airflow yang populer. Untuk referensi otoritatif operator Airflow, baca Referensi Operator dan Hook serta indeks Penyedia.
BashOperator
Gunakan BashOperator untuk menjalankan program command line.
Cloud Composer menjalankan perintah yang diberikan dalam skrip Bash pada pekerja Airflow. Pekerja adalah container Docker berbasis Debian dan mencakup beberapa paket.
PythonOperator
Gunakan PythonOperator untuk menjalankan kode Python arbitrer.
Cloud Composer menjalankan kode Python dalam container yang menyertakan paket untuk versi image Cloud Composer yang digunakan di lingkungan Anda.
Untuk menginstal paket Python tambahan, baca artikel Menginstal Dependensi Python.
Operator Google Cloud
Untuk menjalankan tugas yang menggunakan produk Google Cloud, gunakan operator Google Cloud Airflow. Misalnya, operator BigQuery mengkueri dan memproses data di BigQuery.
Ada banyak operator Airflow lainnya untuk Google Cloud dan masing-masing layanan yang disediakan oleh Google Cloud. Baca bagian Operator Google Cloud untuk mengetahui daftar lengkapnya.
EmailOperator
Gunakan EmailOperator untuk mengirim email dari DAG. Untuk mengirim email dari lingkungan Cloud Composer, Anda harus mengonfigurasi lingkungan Anda untuk menggunakan SendGrid.
Notifikasi tentang kegagalan operator
Tetapkan email_on_failure
ke True
untuk mengirim notifikasi email saat operator dalam DAG gagal. Untuk mengirim notifikasi email dari lingkungan Cloud Composer, Anda harus mengonfigurasi lingkungan untuk menggunakan SendGrid.
Panduan alur kerja DAG
Tempatkan library Python kustom apa pun di arsip ZIP DAG di direktori bertingkat. Jangan menempatkan library di tingkat atas direktori DAG.
Saat Airflow memindai folder
dags/
, Airflow hanya memeriksa DAG di modul Python yang ada di level teratas folder DAG dan di tingkat atas arsip ZIP yang juga terletak di folderdags/
level atas. Jika Airflow menemukan modul Python dalam arsip ZIP yang tidak berisi substringairflow
danDAG
, Airflow akan berhenti memproses arsip ZIP. Airflow hanya menampilkan DAG yang ditemukan hingga saat itu.Untuk fault tolerance, jangan menentukan beberapa objek DAG dalam modul Python yang sama.
Jangan gunakan SubDAG. Sebagai gantinya, kelompokkan tugas di dalam DAG.
Tempatkan file yang diperlukan pada waktu penguraian DAG ke dalam folder
dags/
, bukan di folderdata/
.Uji DAG yang dikembangkan atau diubah seperti yang direkomendasikan dalam petunjuk untuk menguji DAG.
Alat CLI Pengembangan Lokal Compose menyederhanakan pengembangan DAG Apache Airflow untuk Cloud Composer 2 dengan menjalankan lingkungan Airflow secara lokal. Lingkungan Airflow lokal ini menggunakan image versi Cloud Composer 2 tertentu.
Pastikan DAG yang dikembangkan tidak meningkatkan waktu penguraian DAG terlalu banyak.
Tugas Airflow bisa gagal karena beberapa alasan. Untuk menghindari kegagalan seluruh DAG berjalan, sebaiknya aktifkan percobaan ulang tugas. Menyetel percobaan ulang maksimum ke
0
berarti tidak ada percobaan ulang yang dilakukan.Sebaiknya ganti opsi
default_task_retries
dengan nilai untuk menghentikan tugas selain0
. Selain itu, Anda dapat menetapkan parameterretries
pada level tugas.Jika Anda ingin menggunakan GPU dalam tugas Airflow, buat cluster GKE terpisah berdasarkan node menggunakan mesin dengan GPU. Gunakan GKEStartPodOperator untuk menjalankan tugas Anda.
Hindari menjalankan tugas yang menggunakan CPU dan memori berat di kumpulan node cluster tempat komponen Airflow lainnya (penjadwal, pekerja, server web) dijalankan. Sebagai gantinya, gunakan KubernetesPodOperator atau GKEStartPodOperator.
Saat men-deploy DAG ke lingkungan, hanya upload file yang benar-benar diperlukan untuk menafsirkan dan menjalankan DAG ke dalam folder
/dags
.Batasi jumlah file DAG di folder
/dags
.Airflow terus mengurai DAG di folder
/dags
. Penguraian adalah proses yang melakukan loop melalui folder DAG dan jumlah file yang perlu dimuat (dengan dependensinya) berdampak pada performa penguraian DAG dan penjadwalan tugas. Jauh lebih efisien untuk menggunakan 100 file dengan 100 DAG daripada 10.000 file dengan masing-masing 1 DAG, sehingga pengoptimalan seperti itu direkomendasikan. Pengoptimalan ini adalah keseimbangan antara waktu penguraian dan efisiensi pembuatan dan pengelolaan DAG.Anda juga dapat mempertimbangkan, misalnya, untuk men-deploy 10.000 file DAG, Anda dapat membuat 100 file zip yang masing-masing berisi 100 file DAG.
Selain petunjuk di atas, jika Anda memiliki lebih dari 10.000 file DAG, pembuatan DAG dengan cara terprogram mungkin merupakan opsi yang baik. Misalnya, Anda dapat mengimplementasikan satu file DAG Python yang menghasilkan sejumlah objek DAG (misalnya, 20.100 objek DAG).
FAQ untuk menulis DAG
Bagaimana cara meminimalkan pengulangan kode jika saya ingin menjalankan tugas yang sama atau serupa di beberapa DAG?
Sebaiknya tentukan library dan wrapper untuk meminimalkan pengulangan kode.
Bagaimana cara menggunakan kembali kode antar-file DAG?
Tempatkan fungsi utilitas Anda di library Python lokal, lalu impor fungsi tersebut. Anda dapat mereferensikan fungsi ini di DAG mana pun yang berada di folder dags/
di bucket lingkungan Anda.
Bagaimana cara meminimalkan risiko dari perbedaan definisi yang timbul?
Misalnya, Anda memiliki dua tim yang ingin menggabungkan data mentah ke dalam metrik pendapatan. Tim menulis dua tugas yang sedikit berbeda yang menyelesaikan hal yang sama. Tentukan library untuk menangani data pendapatan sehingga pengimplementasi DAG harus mengklarifikasi definisi pendapatan yang digabungkan.
Bagaimana cara menetapkan dependensi antar-DAG?
Ini bergantung pada cara Anda ingin mendefinisikan dependensi.
Jika memiliki dua DAG (DAG A dan DAG B) dan ingin DAG B dipicu setelah DAG
A, Anda dapat menempatkan
TriggerDagRunOperator
di akhir
Dag A.
Jika DAG B hanya bergantung pada artefak yang dihasilkan DAG A, seperti pesan Pub/Sub, sensor mungkin akan berfungsi lebih baik.
Jika DAG B terintegrasi erat dengan DAG A, Anda mungkin dapat menggabungkan kedua DAG menjadi satu DAG.
Bagaimana cara meneruskan ID operasi unik ke DAG dan tugasnya?
Misalnya, Anda ingin meneruskan nama cluster Dataproc dan jalur file.
Anda dapat membuat ID unik acak dengan menampilkan str(uuid.uuid4())
dalam
PythonOperator
. Tindakan ini akan menempatkan ID ke
XComs
,
sehingga Anda dapat merujuk ke ID tersebut di operator lain melalui kolom template.
Sebelum membuat uuid
, pertimbangkan apakah ID khusus DagRun akan
lebih berharga. Anda juga dapat mereferensikan ID ini dalam penggantian Jinja dengan
menggunakan makro.
Bagaimana cara memisahkan tugas di DAG?
Setiap tugas harus menjadi unit kerja idempoten. Akibatnya, Anda harus menghindari
enkapsulasi alur kerja multi-langkah dalam satu tugas, seperti program kompleks
yang berjalan di PythonOperator
.
Haruskah saya menentukan beberapa tugas dalam satu DAG untuk menggabungkan data dari berbagai sumber?
Misalnya, Anda memiliki beberapa tabel dengan data mentah dan ingin membuat agregat harian untuk setiap tabel. Tugas-tugas tersebut tidak saling bergantung. Haruskah Anda membuat satu tugas dan DAG untuk setiap tabel atau membuat satu DAG umum?
Jika Anda tidak keberatan dengan setiap tugas yang menggunakan properti level DAG yang sama, misalnya
schedule_interval
, sebaiknya tentukan beberapa tugas dalam satu
DAG. Jika tidak, untuk meminimalkan pengulangan kode, beberapa DAG dapat dibuat
dari satu modul Python dengan menempatkannya ke dalam globals()
modul.
Bagaimana cara membatasi jumlah tugas serentak yang berjalan di DAG?
Misalnya, Anda tidak ingin melebihi batas penggunaan/kuota API atau tidak menjalankan terlalu banyak proses bersamaan.
Anda dapat menentukan kumpulan Airflow di UI web Airflow dan mengaitkan tugas dengan kumpulan yang ada di DAG.
FAQ terkait penggunaan operator
Haruskah saya menggunakan DockerOperator
?
Sebaiknya jangan gunakan
DockerOperator
,
kecuali jika digunakan untuk meluncurkan container pada penginstalan Docker jarak jauh (bukan
di dalam cluster lingkungan). Di lingkungan Cloud Composer, operator tidak memiliki akses ke daemon Docker.
Sebagai gantinya, gunakan KubernetesPodOperator
atau
GKEStartPodOperator
. Operator ini masing-masing meluncurkan pod Kubernetes ke
cluster Kubernetes atau GKE. Perlu diperhatikan bahwa kami tidak
merekomendasikan peluncuran pod ke cluster lingkungan karena hal ini dapat menyebabkan
persaingan resource.
Haruskah saya menggunakan SubDagOperator
?
Sebaiknya jangan gunakan SubDagOperator
.
Gunakan alternatif seperti yang disarankan dalam Petunjuk Mengelompokkan Tugas.
Haruskah saya menjalankan kode Python hanya dalam PythonOperators
untuk memisahkan operator Python sepenuhnya?
Bergantung pada sasaran, Anda memiliki beberapa opsi.
Jika satu-satunya kekhawatiran Anda adalah mempertahankan dependensi Python terpisah, Anda
dapat menggunakan PythonVirtualenvOperator
.
Sebaiknya gunakan KubernetesPodOperator
. Operator ini dapat Anda gunakan untuk menentukan pod Kubernetes dan menjalankan pod tersebut di cluster lain.
Bagaimana cara menambahkan paket biner atau non-PyPI kustom?
Anda dapat menginstal paket yang dihosting di repositori paket pribadi.
Anda juga dapat menggunakan KubernetesPodOperator
untuk menjalankan pod Kubernetes dengan image Anda sendiri yang dibangun dengan paket kustom.
Bagaimana cara meneruskan argumen secara seragam ke DAG dan tugasnya?
Anda dapat menggunakan dukungan bawaan Airflow untuk Pembuatan template Jinja guna meneruskan argumen yang dapat digunakan di kolom dengan template.
Kapan penggantian template terjadi?
Penggantian template terjadi pada pekerja Airflow tepat sebelum fungsi pre_execute
operator dipanggil. Dalam praktiknya, ini berarti template
tidak diganti sampai tepat sebelum tugas berjalan.
Bagaimana cara mengetahui argumen operator mana yang mendukung penggantian template?
Argumen operator yang mendukung penggantian template Jinja2 secara eksplisit ditandai seperti itu.
Cari kolom template_fields
di definisi Operator, yang berisi daftar nama argumen yang mengalami penggantian template.
Misalnya, lihat
BashOperator
,
yang mendukung pemberian template untuk argumen bash_command
dan env
.
Langkah selanjutnya
- Memecahkan masalah DAG
- Memecahkan Masalah Penjadwal
- Operator Google
- Operator Google Cloud
- Tutorial Apache Airflow
- Antarmuka Publik Airflow
- Operator Core Airflow di GitHub. Lihat cabang untuk rilis Airflow Anda.
- Operator paket penyedia di GitHub. Lihat cabang untuk rilis Airflow Anda.