Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3
Panduan ini menunjukkan cara menulis directed acyclic graph (DAG) Apache Airflow yang berjalan di lingkungan Cloud Composer.
Karena Apache Airflow tidak menyediakan isolasi tugas dan DAG yang kuat, sebaiknya Anda menggunakan lingkungan produksi dan pengujian terpisah untuk mencegah gangguan DAG. Untuk informasi selengkapnya, lihat Menguji DAG.
Membuat struktur DAG Airflow
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 Alur Kerja:
Operator dan tugas
Operator Airflow menjelaskan pekerjaan yang akan dilakukan. Tugas tugas adalah instance operator tertentu.
Hubungan tugas
Hubungan tugas menjelaskan urutan pekerjaan yang harus diselesaikan.
Contoh alur kerja DAG lengkap di Python
Alur kerja berikut adalah template DAG lengkap yang berfungsi dan terdiri dari
dua tugas: tugas hello_python
dan tugas goodbye_bash
:
Untuk informasi selengkapnya tentang menentukan DAG Airflow, lihat tutorial Airflow dan konsep Airflow.
Operator Airflow
Contoh berikut menunjukkan beberapa operator Airflow populer. Untuk referensi resmi operator Airflow, lihat Referensi Operator dan Hook dan Indeks penyedia.
BashOperator
Gunakan BashOperator untuk menjalankan program command line.
Cloud Composer menjalankan perintah yang disediakan dalam skrip Bash di pekerja Airflow. Pekerja adalah penampung Docker berbasis Debian dan menyertakan beberapa paket.
- Perintah
gcloud
, termasuk sub-perintahgcloud storage
untuk menggunakan bucket Cloud Storage. - Perintah
bq
- Perintah
kubectl
PythonOperator
Gunakan PythonOperator untuk menjalankan kode Python arbitrer.
Cloud Composer menjalankan kode Python dalam penampung yang menyertakan paket untuk versi image Cloud Composer yang digunakan di lingkungan Anda.
Untuk menginstal paket Python tambahan, lihat Menginstal Dependensi Python.
Operator Google Cloud
Untuk menjalankan tugas yang menggunakan produk Google Cloud, gunakan operator Google Cloud Airflow. Misalnya, operator BigQuery membuat kueri dan memproses data di BigQuery.
Ada banyak lagi operator Airflow untuk Google Cloud dan layanan individual yang disediakan oleh Google Cloud. Lihat Operator Google Cloud untuk mengetahui daftar lengkapnya.
EmailOperator
Gunakan EmailOperator untuk mengirim email dari DAG. Untuk mengirim email dari lingkungan Cloud Composer, konfigurasi 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 di arsip ZIP DAG dalam direktori bertingkat. Jangan menempatkan library di tingkat teratas direktori DAG.
Saat memindai folder
dags/
, Airflow hanya memeriksa DAG di modul Python yang berada di tingkat atas folder DAG dan di tingkat atas arsip ZIP yang juga terletak di folderdags/
tingkat 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 titik tersebut.Untuk toleransi error, jangan tentukan 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 folder
dags/
, bukan di folderdata/
.Uji DAG yang dikembangkan atau diubah seperti yang direkomendasikan dalam petunjuk untuk menguji DAG.
Pastikan DAG yang dikembangkan tidak terlalu meningkatkan waktu penguraian DAG.
Tugas Airflow dapat gagal karena beberapa alasan. Untuk menghindari kegagalan seluruh DAG yang berjalan, sebaiknya aktifkan percobaan ulang tugas. Menetapkan percobaan ulang maksimum ke
0
berarti tidak ada percobaan ulang yang dilakukan.Sebaiknya ganti opsi
default_task_retries
dengan nilai untuk percobaan ulang tugas selain0
. Selain itu, Anda dapat menetapkan parameterretries
di tingkat tugas.Jika Anda ingin menggunakan GPU dalam tugas Airflow, buat cluster GKE terpisah berdasarkan node yang menggunakan mesin dengan GPU. Gunakan GKEStartPodOperator untuk menjalankan tugas Anda.
Hindari menjalankan tugas yang menggunakan banyak CPU dan memori di node pool cluster tempat komponen Airflow lainnya (penjadwal, pekerja, server web) berjalan. Sebagai gantinya, gunakan KubernetesPodOperator atau GKEStartPodOperator.
Saat men-deploy DAG ke lingkungan, hanya upload file yang benar-benar diperlukan untuk menafsirkan dan mengeksekusi DAG ke folder
/dags
.Batasi jumlah file DAG di folder
/dags
.Airflow terus mengurai DAG di folder
/dags
. Mengurai adalah proses yang melakukan loop melalui folder DAG dan jumlah file yang perlu dimuat (dengan dependensinya) akan memengaruhi performa penguraian DAG dan penjadwalan tugas. Sebaiknya gunakan 100 file dengan masing-masing 100 DAG daripada 10.000 file dengan masing-masing 1 DAG karena pengoptimalan tersebut jauh lebih efisien. Pengoptimalan ini adalah keseimbangan antara waktu penguraian dan efisiensi pembuatan serta 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, membuat DAG secara terprogram mungkin merupakan opsi yang baik. Misalnya, Anda dapat menerapkan satu file DAG Python yang menghasilkan sejumlah objek DAG (misalnya, 20, 100 objek DAG).
Hindari penggunaan operator Airflow yang tidak digunakan lagi. Sebagai gantinya, gunakan alternatif terbarunya.
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 di antara file DAG?
Masukkan fungsi utilitas Anda di library Python lokal dan impor fungsi tersebut. Anda dapat mereferensikan fungsi dalam DAG apa pun yang terletak di folder dags/
di bucket lingkungan Anda.
Bagaimana cara meminimalkan risiko munculnya definisi yang berbeda?
Misalnya, Anda memiliki dua tim yang ingin menggabungkan data mentah menjadi metrik pendapatan. Tim menulis dua tugas yang sedikit berbeda yang mencapai hal yang sama. Menentukan library untuk menggunakan data pendapatan sehingga pengimplementasi DAG harus mengklarifikasi definisi pendapatan yang digabungkan.
Bagaimana cara menetapkan dependensi antar-DAG?
Hal ini bergantung pada cara Anda ingin menentukan 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 run unik ke DAG dan tugasnya?
Misalnya, Anda ingin meneruskan nama cluster dan jalur file Dataproc.
Anda dapat membuat ID unik acak dengan menampilkan str(uuid.uuid4())
dalam
PythonOperator
. Tindakan ini akan menempatkan ID ke dalam
XComs
sehingga Anda dapat merujuk ke ID di operator lain
melalui kolom template.
Sebelum membuat uuid
, pertimbangkan apakah ID khusus DagRun akan lebih bernilai. Anda juga dapat mereferensikan ID ini dalam penggantian Jinja dengan menggunakan makro.
Bagaimana cara memisahkan tugas dalam DAG?
Setiap tugas harus berupa unit pekerjaan idempoten. Oleh karena itu, 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 beberapa 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 jika setiap tugas memiliki properti tingkat DAG yang sama, seperti
schedule_interval
, sebaiknya tentukan beberapa tugas dalam satu
DAG. Atau, untuk meminimalkan pengulangan kode, beberapa DAG dapat dihasilkan
dari satu modul Python dengan menempatkannya ke dalam globals()
modul.
Bagaimana cara membatasi jumlah tugas serentak yang berjalan di DAG?
Misalnya, Anda ingin menghindari melampaui batas/kuota penggunaan API atau menghindari menjalankan terlalu banyak proses serentak.
Anda dapat menentukan kumpulan Airflow di UI web Airflow dan mengaitkan tugas dengan kumpulan yang ada di DAG.
FAQ untuk menggunakan operator
Haruskah saya menggunakan DockerOperator
?
Sebaiknya jangan gunakan
DockerOperator
, kecuali jika digunakan untuk meluncurkan
penampung di penginstalan Docker jarak jauh (bukan dalam cluster
lingkungan). Di lingkungan Cloud Composer, operator tidak memiliki
akses ke daemon Docker.
Sebagai gantinya, gunakan KubernetesPodOperator
atau
GKEStartPodOperator
. Operator ini meluncurkan pod Kubernetes ke dalam
cluster Kubernetes atau GKE. Perhatikan bahwa kami tidak
menyarankan untuk meluncurkan pod ke cluster lingkungan, karena hal ini dapat menyebabkan
persaingan resource.
Haruskah saya menggunakan SubDagOperator
?
Sebaiknya jangan gunakan SubDagOperator
.
Gunakan alternatif seperti yang disarankan di Mengelompokkan tugas.
Haruskah saya menjalankan kode Python hanya di PythonOperators
untuk memisahkan operator Python sepenuhnya?
Bergantung pada sasaran Anda, Anda memiliki beberapa opsi.
Jika satu-satunya masalah Anda adalah mempertahankan dependensi Python terpisah, Anda
dapat menggunakan PythonVirtualenvOperator
.
Pertimbangkan untuk menggunakan KubernetesPodOperator
. Operator ini memungkinkan Anda
menentukan pod Kubernetes dan menjalankan pod di cluster lain.
Bagaimana cara menambahkan paket biner kustom atau non-PyPI?
Anda dapat menginstal paket yang dihosting di repositori paket pribadi.
Bagaimana cara meneruskan argumen secara seragam ke DAG dan tugasnya?
Anda dapat menggunakan dukungan bawaan Airflow untuk template Jinja guna meneruskan argumen yang dapat digunakan di kolom 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 hingga tepat sebelum tugas berjalan.
Bagaimana cara mengetahui argumen operator yang mendukung penggantian template?
Argumen operator yang mendukung penggantian template Jinja2 ditandai secara eksplisit sebagai argumen tersebut.
Cari kolom template_fields
dalam definisi Operator,
yang berisi daftar nama argumen yang mengalami penggantian template.
Misalnya, lihat
BashOperator
, yang mendukung pembuatan template untuk
argumen bash_command
dan env
.
Operator Airflow yang tidak digunakan lagi dan dihapus
Operator Airflow yang tercantum dalam tabel berikut tidak digunakan lagi:
Hindari penggunaan operator ini di DAG Anda. Sebagai gantinya, gunakan operator pengganti terbaru yang disediakan.
Jika operator tercantum sebagai dihapus, berarti operator tersebut sudah tidak tersedia di salah satu build Airflow yang dirilis di Cloud Composer 3.
Jika operator tercantum sebagai direncanakan untuk dihapus, operator tersebut tidak digunakan lagi dan akan dihapus dalam build Airflow mendatang.
Operator yang tidak digunakan lagi | Status | Operator pengganti | Penggantian tersedia dari |
---|---|---|---|
CreateAutoMLTextTrainingJobOperator | Dihapus | SupervisedFineTuningTrainOperator |
composer-3-airflow-2.9.3-build.1 composer-3-airflow-2.9.1-build.8 |
GKEDeploymentHook | Dihapus | GKEKubernetesHook |
Semua versi |
GKECustomResourceHook | Dihapus | GKEKubernetesHook |
Semua versi |
GKEPodHook | Dihapus | GKEKubernetesHook |
Semua versi |
GKEJobHook | Dihapus | GKEKubernetesHook |
Semua versi |
GKEPodAsyncHook | Dihapus | GKEKubernetesAsyncHook |
Semua versi |
SecretsManagerHook | Dihapus | GoogleCloudSecretManagerHook |
composer-3-airflow-2.7.3-build.6 |
BigQueryExecuteQueryOperator | Dihapus | BigQueryInsertJobOperator |
Semua versi |
BigQueryPatchDatasetOperator | Dihapus | BigQueryUpdateDatasetOperator |
Semua versi |
DataflowCreateJavaJobOperator | Dihapus | beam.BeamRunJavaPipelineOperator |
Semua versi |
DataflowCreatePythonJobOperator | Dihapus | beam.BeamRunPythonPipelineOperator |
Semua versi |
DataprocSubmitPigJobOperator | Dihapus | DataprocSubmitJobOperator |
Semua versi |
DataprocSubmitHiveJobOperator | Dihapus | DataprocSubmitJobOperator |
Semua versi |
DataprocSubmitSparkSqlJobOperator | Dihapus | DataprocSubmitJobOperator |
Semua versi |
DataprocSubmitSparkJobOperator | Dihapus | DataprocSubmitJobOperator |
Semua versi |
DataprocSubmitHadoopJobOperator | Dihapus | DataprocSubmitJobOperator |
Semua versi |
DataprocSubmitPySparkJobOperator | Dihapus | DataprocSubmitJobOperator |
Semua versi |
BigQueryTableExistenceAsyncSensor | Dihapus | BigQueryTableExistenceSensor |
Semua versi |
BigQueryTableExistencePartitionAsyncSensor | Dihapus | BigQueryTablePartitionExistenceSensor |
Semua versi |
CloudComposerEnvironmentSensor | Dihapus | CloudComposerCreateEnvironmentOperator, CloudComposerDeleteEnvironmentOperator, CloudComposerUpdateEnvironmentOperator |
Semua versi |
GCSObjectExistenceAsyncSensor | Dihapus | GCSObjectExistenceSensor |
Semua versi |
GoogleAnalyticsHook | Dihapus | GoogleAnalyticsAdminHook |
Semua versi |
GoogleAnalyticsListAccountsOperator | Dihapus | GoogleAnalyticsAdminListAccountsOperator |
Semua versi |
GoogleAnalyticsGetAdsLinkOperator | Dihapus | GoogleAnalyticsAdminGetGoogleAdsLinkOperator |
Semua versi |
GoogleAnalyticsRetrieveAdsLinksListOperator | Dihapus | GoogleAnalyticsAdminListGoogleAdsLinksOperator |
Semua versi |
GoogleAnalyticsDataImportUploadOperator | Dihapus | GoogleAnalyticsAdminCreateDataStreamOperator |
Semua versi |
GoogleAnalyticsDeletePreviousDataUploadsOperator | Dihapus | GoogleAnalyticsAdminDeleteDataStreamOperator |
Semua versi |
DataPipelineHook | Tidak digunakan lagi, penghapusan direncanakan | DataflowHook |
Akan diumumkan |
CreateDataPipelineOperator | Tidak digunakan lagi, penghapusan direncanakan | DataflowCreatePipelineOperator |
Akan diumumkan |
RunDataPipelineOperator | Tidak digunakan lagi, penghapusan direncanakan | DataflowRunPipelineOperator |
Akan diumumkan |
AutoMLDatasetLink | Tidak digunakan lagi, penghapusan direncanakan | TranslationLegacyDatasetLink |
Akan diumumkan |
AutoMLDatasetListLink | Tidak digunakan lagi, penghapusan direncanakan | TranslationDatasetListLink |
Akan diumumkan |
AutoMLModelLink | Tidak digunakan lagi, penghapusan direncanakan | TranslationLegacyModelLink |
Akan diumumkan |
AutoMLModelTrainLink | Tidak digunakan lagi, penghapusan direncanakan | TranslationLegacyModelTrainLink |
Akan diumumkan |
AutoMLModelPredictLink | Tidak digunakan lagi, penghapusan direncanakan | TranslationLegacyModelPredictLink |
Akan diumumkan |
AutoMLBatchPredictOperator | Tidak digunakan lagi, penghapusan direncanakan | vertex_ai.batch_prediction_job |
Akan diumumkan |
AutoMLPredictOperator | Tidak digunakan lagi, penghapusan direncanakan | vertex_aigenerative_model. TextGenerationModelPredictOperator, translate.TranslateTextOperator |
Akan diumumkan |
PromptLanguageModelOperator | Tidak digunakan lagi, penghapusan direncanakan | TextGenerationModelPredictOperator |
Akan diumumkan |
GenerateTextEmbeddingsOperator | Tidak digunakan lagi, penghapusan direncanakan | TextEmbeddingModelGetEmbeddingsOperator |
Akan diumumkan |
PromptMultimodalModelOperator | Tidak digunakan lagi, penghapusan direncanakan | GenerativeModelGenerateContentOperator |
Akan diumumkan |
PromptMultimodalModelWithMediaOperator | Tidak digunakan lagi, penghapusan direncanakan | GenerativeModelGenerateContentOperator |
Akan diumumkan |
DataflowStartSqlJobOperator | Tidak digunakan lagi, penghapusan direncanakan | DataflowStartYamlJobOperator |
Akan diumumkan |
LifeSciencesHook | Tidak digunakan lagi, penghapusan direncanakan | Hook Operator Batch Google Cloud |
Akan diumumkan |
DataprocScaleClusterOperator | Tidak digunakan lagi, penghapusan direncanakan | DataprocUpdateClusterOperator |
Akan diumumkan |
MLEngineStartBatchPredictionJobOperator | Tidak digunakan lagi, penghapusan direncanakan | CreateBatchPredictionJobOperator |
Akan diumumkan |
MLEngineManageModelOperator | Tidak digunakan lagi, penghapusan direncanakan | MLEngineCreateModelOperator, MLEngineGetModelOperator |
Akan diumumkan |
MLEngineGetModelOperator | Tidak digunakan lagi, penghapusan direncanakan | GetModelOperator |
Akan diumumkan |
MLEngineDeleteModelOperator | Tidak digunakan lagi, penghapusan direncanakan | DeleteModelOperator |
Akan diumumkan |
MLEngineManageVersionOperator | Tidak digunakan lagi, penghapusan direncanakan | MLEngineCreateVersion, MLEngineSetDefaultVersion, MLEngineListVersions, MLEngineDeleteVersion |
Akan diumumkan |
MLEngineCreateVersionOperator | Tidak digunakan lagi, penghapusan direncanakan | Parameter parent_model untuk operator VertexAI |
Akan diumumkan |
MLEngineSetDefaultVersionOperator | Tidak digunakan lagi, penghapusan direncanakan | SetDefaultVersionOnModelOperator |
Akan diumumkan |
MLEngineListVersionsOperator | Tidak digunakan lagi, penghapusan direncanakan | ListModelVersionsOperator |
Akan diumumkan |
MLEngineDeleteVersionOperator | Tidak digunakan lagi, penghapusan direncanakan | DeleteModelVersionOperator |
Akan diumumkan |
MLEngineStartTrainingJobOperator | Tidak digunakan lagi, penghapusan direncanakan | CreateCustomPythonPackageTrainingJobOperator |
Akan diumumkan |
MLEngineTrainingCancelJobOperator | Tidak digunakan lagi, penghapusan direncanakan | CancelCustomTrainingJobOperator |
Akan diumumkan |
LifeSciencesRunPipelineOperator | Tidak digunakan lagi, penghapusan direncanakan | Operator Batch Google Cloud |
Akan diumumkan |
MLEngineCreateModelOperator | Tidak digunakan lagi, penghapusan direncanakan | operator VertexAI yang sesuai |
Akan diumumkan |
Langkah selanjutnya
- Memecahkan masalah DAG
- Memecahkan masalah Penjadwal
- Operator Google
- Operator Google Cloud
- Tutorial Apache Airflow