Memecahkan masalah penjadwal Airflow

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

Halaman ini memberikan langkah-langkah dan informasi pemecahan masalah untuk masalah umum terkait scheduler Airflow dan pemroses DAG.

Mengidentifikasi sumber masalah

Untuk memulai pemecahan masalah, identifikasi apakah masalah terjadi:

  • Pada waktu penguraian DAG, saat DAG diuraikan oleh pemroses DAG Airflow
  • Pada waktu eksekusi, saat DAG diproses oleh scheduler Airflow

Untuk mengetahui informasi selengkapnya tentang waktu penguraian dan waktu eksekusi, baca Perbedaan antara waktu penguraian DAG dan waktu eksekusi DAG.

Memeriksa masalah pemrosesan DAG

  1. Periksa log pemroses DAG.
  2. Periksa waktu penguraian DAG.

Memantau tugas yang berjalan dan dalam antrean

Untuk memeriksa apakah ada tugas yang stuck dalam antrean, ikuti langkah-langkah berikut.

  1. Di Google Cloud konsol, buka halaman Environments.

    Buka Environments

  2. Di daftar lingkungan, klik nama lingkungan Anda. Halaman Environment details akan terbuka.

  3. Buka tab Monitoring.

  4. Di tab Monitoring, tinjau diagram Airflow tasks chart di bagian DAG runs dan identifikasi kemungkinan masalah. Tugas Airflow adalah tugas yang berada dalam status antrean di Airflow, dan dapat masuk ke antrean broker Celery atau Kubernetes Executor. Tugas yang diantrekan Celery adalah instance tugas yang dimasukkan ke antrean broker Celery.

Memecahkan masalah pada waktu penguraian DAG

Bagian berikut menjelaskan gejala dan potensi perbaikan untuk beberapa masalah umum pada waktu penguraian DAG.

Penguraian dan penjadwalan DAG di Managed Airflow (Gen 1 Lama) dan Airflow 1

Efisiensi penguraian DAG ditingkatkan secara signifikan di Airflow 2. Jika Anda mengalami masalah performa terkait penguraian dan penjadwalan DAG, pertimbangkan untuk bermigrasi ke Airflow 2.

Di Managed Airflow (Gen 1 Lama), scheduler berjalan di node cluster bersama dengan komponen Managed Airflow lainnya. Oleh karena itu, beban node cluster individual mungkin lebih tinggi atau lebih rendah dibandingkan dengan node lainnya. Performa scheduler (penguraian dan penjadwalan DAG) dapat bervariasi bergantung pada node tempat scheduler berjalan. Selain itu, node individual tempat scheduler berjalan dapat berubah sebagai hasil dari operasi upgrade atau pemeliharaan. Batasan ini diatasi di Managed Airflow (Gen 2) tempat Anda dapat mengalokasikan resource CPU dan memori ke scheduler, dan performa scheduler tidak bergantung pada beban node cluster.

Jumlah dan distribusi waktu tugas

Airflow dapat mengalami masalah saat menjadwalkan sejumlah besar DAG atau tugas pada waktu yang sama. Untuk menghindari masalah penjadwalan, Anda dapat:

  • Menyesuaikan DAG untuk menggunakan tugas yang lebih sedikit dan lebih terkonsolidasi.
  • Menyesuaikan interval jadwal DAG untuk mendistribusikan operasi DAG secara lebih merata dari waktu ke waktu.

Menskalakan konfigurasi Airflow

Airflow menyediakan opsi konfigurasi Airflow yang mengontrol jumlah tugas dan DAG yang dapat dijalankan Airflow pada waktu yang sama. Untuk menetapkan opsi konfigurasi ini, ganti nilainya untuk lingkungan Anda. Anda juga dapat menetapkan beberapa nilai ini di tingkat DAG atau tugas.

  • Konkurensi worker

    Parameter [celery]worker_concurrency mengontrol jumlah maksimum tugas yang dapat dijalankan oleh worker Airflow pada waktu yang sama. Jika Anda mengalikan nilai parameter ini dengan jumlah worker Airflow di lingkungan Managed Airflow, Anda akan mendapatkan jumlah maksimum tugas yang dapat dijalankan pada waktu tertentu di lingkungan Anda. Jumlah ini dibatasi oleh opsi konfigurasi Airflow [core]parallelism, yang dijelaskan lebih lanjut.

  • Operasi DAG aktif maksimum

    Opsi konfigurasi Airflow [core]max_active_runs_per_dag mengontrol jumlah maksimum operasi DAG aktif per DAG. Scheduler tidak membuat lebih banyak operasi DAG jika mencapai batas ini.

    Jika parameter ini ditetapkan dengan tidak benar, Anda mungkin mengalami masalah saat scheduler membatasi eksekusi DAG karena tidak dapat membuat lebih banyak instance operasi DAG pada waktu tertentu.

    Anda juga dapat menetapkan nilai ini di tingkat DAG dengan parameter max_active_runs.

  • Tugas aktif maksimum per DAG

    Opsi konfigurasi Airflow [core]max_active_tasks_per_dag mengontrol jumlah maksimum instance tugas yang dapat berjalan secara serentak di setiap DAG.

    Jika parameter ini ditetapkan dengan tidak benar, Anda mungkin mengalami masalah saat eksekusi satu instance DAG berjalan lambat karena hanya ada jumlah tugas DAG terbatas yang dapat dijalankan pada waktu tertentu. Dalam hal ini, Anda dapat meningkatkan nilai opsi konfigurasi ini.

    Anda juga dapat menetapkan nilai ini di tingkat DAG dengan parameter max_active_tasks.

    Anda dapat menggunakan max_active_tis_per_dag dan max_active_tis_per_dagrun parameter di tingkat tugas untuk mengontrol jumlah instance dengan ID tugas tertentu yang diizinkan untuk berjalan per DAG dan per operasi DAG.

  • Paralelisme dan ukuran pool

    Opsi konfigurasi Airflow [core]parallelism mengontrol jumlah tugas yang dapat diantrekan oleh scheduler Airflow dalam antrean Executor setelah semua dependensi untuk tugas ini terpenuhi.

    Ini adalah parameter global untuk seluruh penyiapan Airflow.

    Tugas diantrekan dan dijalankan dalam pool. Lingkungan Managed Airflow hanya menggunakan satu pool. Ukuran pool ini mengontrol jumlah tugas yang dapat diantrekan oleh scheduler untuk dieksekusi pada waktu tertentu. Jika ukuran pool terlalu kecil, scheduler tidak dapat mengantrekan tugas untuk dieksekusi meskipun nilai minimum, yang ditentukan oleh opsi konfigurasi [core]parallelism dan oleh opsi konfigurasi [celery]worker_concurrency yang dikalikan dengan jumlah worker Airflow, belum terpenuhi.

    Anda dapat mengonfigurasi ukuran pool di UI Airflow (Admin > Pools). Sesuaikan ukuran pool dengan tingkat paralelisme yang Anda harapkan di lingkungan Anda.

    Biasanya, [core]parallelism ditetapkan sebagai produk dari jumlah maksimum worker dan [celery]worker_concurrency.

Memecahkan masalah terkait tugas yang berjalan dan dalam antrean

Bagian berikut menjelaskan gejala dan potensi perbaikan untuk beberapa masalah umum terkait tugas yang berjalan dan dalam antrean.

Operasi DAG tidak dijalankan

Gejala:

Jika tanggal jadwal untuk DAG ditetapkan secara dinamis, hal ini dapat menyebabkan berbagai efek samping yang tidak terduga. Contoh:

  • Eksekusi DAG selalu di masa mendatang, dan DAG tidak pernah dieksekusi.

  • Operasi DAG sebelumnya ditandai sebagai dieksekusi dan berhasil meskipun tidak dieksekusi.

Informasi selengkapnya tersedia di dokumentasi Apache Airflow.

Kemungkinan solusi:

  • Ikuti rekomendasi dalam dokumentasi Apache Airflow.

  • Tetapkan start_date statis untuk DAG. Sebagai opsi, Anda dapat menggunakan catchup=False untuk menonaktifkan DAG yang berjalan untuk tanggal sebelumnya.

  • Hindari penggunaan datetime.now() atau days_ago(<number of days>) kecuali jika Anda mengetahui efek samping dari pendekatan ini.

Menggunakan fitur TimeTable scheduler Airflow

Tabel waktu tersedia mulai dari Airflow 2.2.

Anda dapat menentukan tabel waktu untuk DAG dengan salah satu metode berikut:

Anda juga dapat menggunakan Tabel Waktu Bawaan.

Resource cluster terbatas

Anda mungkin mengalami masalah performa jika cluster GKE lingkungan Anda terlalu kecil untuk menangani semua DAG dan tugas Anda. Dalam hal ini, coba salah satu solusi berikut:

  • Buat lingkungan baru dengan jenis mesin yang memberikan performa lebih baik dan migrasikan DAG Anda ke lingkungan tersebut.
  • Buat lebih banyak lingkungan Managed Airflow dan bagi DAG di antara lingkungan tersebut.
  • Ubah jenis mesin untuk node GKE, seperti yang dijelaskan dalam Mengupgrade jenis mesin untuk node GKE. Karena prosedur ini rentan error, prosedur ini adalah opsi yang paling tidak direkomendasikan.
  • Upgrade jenis mesin instance Cloud SQL yang menjalankan database Airflow di lingkungan Anda, misalnya menggunakan gcloud composer environments update perintah. Performa database Airflow yang rendah mungkin menjadi alasan scheduler berjalan lambat.

Menghindari penjadwalan tugas selama masa pemeliharaan

Anda dapat menentukan masa pemeliharaan untuk lingkungan Anda sehingga pemeliharaan lingkungan terjadi di luar waktu saat Anda menjalankan DAG. Anda masih dapat menjalankan DAG selama masa pemeliharaan, selama dapat diterima bahwa beberapa tugas dapat terganggu dan dicoba lagi. Untuk mengetahui informasi selengkapnya tentang pengaruh masa pemeliharaan terhadap lingkungan Anda, lihat Menentukan masa pemeliharaan.

Penggunaan 'wait_for_downstream' di DAG Anda

Jika Anda menetapkan parameter wait_for_downstream ke True di DAG, agar tugas berhasil, semua tugas yang langsung downstream dari tugas ini juga harus berhasil. Artinya, eksekusi tugas yang termasuk dalam operasi DAG tertentu mungkin diperlambat oleh eksekusi tugas dari operasi DAG sebelumnya. Baca selengkapnya di dokumentasi Airflow.

Tugas yang diantrekan terlalu lama akan dibatalkan dan dijadwalkan ulang

Jika tugas Airflow disimpan dalam antrean terlalu lama, scheduler akan menjadwalkan ulang tugas tersebut untuk dieksekusi setelah jumlah waktu yang ditetapkan dalam opsi konfigurasi Airflow [scheduler]task_queued_timeout berlalu. Nilai defaultnya adalah 2400. Di versi Airflow sebelum 2.3.1, tugas juga ditandai sebagai gagal dan dicoba lagi jika memenuhi syarat untuk dicoba lagi.

Salah satu cara untuk mengamati gejala situasi ini adalah dengan melihat diagram yang berisi jumlah tugas yang diantrekan (tab "Monitoring" di UI Managed Airflow). Jika lonjakan dalam diagram ini tidak turun dalam waktu sekitar dua jam, tugas kemungkinan besar akan dijadwalkan ulang (tanpa log), diikuti oleh entri log "Adopted tasks were still pending ..." di log scheduler. Dalam kasus seperti itu, Anda mungkin melihat pesan "Log file is not found..." di log tugas Airflow karena tugas tidak dieksekusi.

Secara umum, perilaku ini diharapkan dan instance tugas terjadwal berikutnya dimaksudkan untuk dieksekusi sesuai jadwal. Jika Anda mengamati banyak kasus seperti itu di lingkungan Managed Airflow, hal ini mungkin berarti tidak ada cukup worker Airflow di lingkungan Anda untuk memproses semua tugas terjadwal.

Resolusi: Untuk mengatasi masalah ini, Anda harus memastikan selalu ada kapasitas di worker Airflow untuk menjalankan tugas yang diantrekan. Misalnya, Anda dapat meningkatkan jumlah worker atau worker_concurrency. Anda juga dapat menyesuaikan paralelisme atau pool untuk mencegah tugas yang diantrekan melebihi kapasitas yang Anda miliki.

Pendekatan Managed Airflow terhadap parameter min_file_process_interval

Managed Airflow mengubah cara [scheduler]min_file_process_interval digunakan oleh scheduler Airflow.

Airflow 1

Jika Managed Airflow menggunakan Airflow 1, pengguna dapat menetapkan nilai [scheduler]min_file_process_interval antara 0 dan 600 detik. Nilai yang lebih tinggi dari 600 detik akan memberikan hasil yang sama seperti jika [scheduler]min_file_process_interval ditetapkan ke 600 detik.

Airflow 2

Di versi Managed Airflow sebelum 1.19.9, [scheduler]min_file_process_interval diabaikan.

Versi Managed Airflow setelah 1.19.9:

Scheduler Airflow dimulai ulang setelah beberapa kali semua DAG dijadwalkan dan parameter [scheduler]num_runs mengontrol berapa kali hal ini dilakukan oleh scheduler. Saat scheduler mencapai loop penjadwalan [scheduler]num_runs, scheduler akan dimulai ulang. Scheduler adalah komponen stateless dan memulai ulang seperti itu adalah mekanisme pemulihan otomatis untuk masalah apa pun yang mungkin dialami scheduler. Nilai default [scheduler]num_runs adalah 5000.

[scheduler]min_file_process_interval dapat digunakan untuk mengonfigurasi frekuensi penguraian DAG, tetapi parameter ini tidak boleh lebih lama dari waktu yang diperlukan scheduler untuk melakukan loop [scheduler]num_runs saat menjadwalkan DAG Anda.

Menandai tugas sebagai gagal setelah mencapai dagrun_timeout

Scheduler menandai tugas yang belum selesai (berjalan, dijadwalkan, dan diantrekan) sebagai gagal jika operasi DAG tidak selesai dalam dagrun_timeout (parameter DAG).

Solusi:

Gejala database Airflow yang mengalami beban berat

Terkadang, di log scheduler Airflow, Anda mungkin melihat entri log peringatan berikut:

Scheduler heartbeat got an exception: (_mysql_exceptions.OperationalError) (2006, "Lost connection to MySQL server at 'reading initial communication packet', system error: 0")"

Gejala serupa juga dapat diamati di log worker Airflow:

Untuk MySQL:

(_mysql_exceptions.OperationalError) (2006, "Lost connection to MySQL server at
'reading initial communication packet', system error: 0")"

Untuk PostgreSQL:

psycopg2.OperationalError: connection to server at ... failed

Error atau peringatan tersebut mungkin merupakan gejala database Airflow yang kelebihan beban karena jumlah koneksi terbuka atau jumlah kueri yang dieksekusi pada waktu yang sama, baik oleh scheduler maupun oleh komponen Airflow lainnya seperti worker, pemicu, dan server web.

Kemungkinan solusi:

Server web menampilkan peringatan 'The scheduler does not appear to be running'

Scheduler melaporkan heartbeat-nya secara rutin ke database Airflow. Berdasarkan informasi ini, server web Airflow menentukan apakah scheduler aktif.

Terkadang, jika scheduler mengalami beban berat, scheduler mungkin tidak dapat melaporkan heartbeat-nya setiap [scheduler]scheduler_heartbeat_sec.

Dalam situasi seperti itu, server web Airflow mungkin menampilkan peringatan berikut:

The scheduler does not appear to be running. Last heartbeat was received <X>
seconds ago.

Kemungkinan solusi:

  • Tingkatkan resource CPU dan memori untuk scheduler.

  • Optimalkan DAG Anda sehingga penguraian dan penjadwalannya lebih cepat dan tidak menggunakan terlalu banyak resource scheduler.

  • Hindari penggunaan variabel global di DAG Airflow. Sebagai gantinya, gunakan variabel lingkungan dan variabel Airflow.

  • Tingkatkan nilai opsi konfigurasi Airflow [scheduler]scheduler_health_check_threshold, sehingga server web menunggu lebih lama sebelum melaporkan ketidaktersediaan scheduler.

Solusi untuk masalah yang terjadi selama pengisian ulang DAG

Terkadang, Anda mungkin ingin menjalankan ulang DAG yang sudah dieksekusi. Anda dapat melakukannya dengan perintah CLI Airflow dengan cara berikut:

Airflow 2 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location LOCATION \
   dags backfill -- -B \
   -s START_DATE \
   -e END_DATE \
   DAG_NAME

Untuk menjalankan ulang tugas yang gagal saja untuk DAG tertentu, gunakan juga argumen --rerun-failed-tasks.

Airflow 1 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location LOCATION \
  backfill -- -B \
  -s START_DATE \
  -e END_DATE \
  DAG_NAME

Untuk menjalankan ulang tugas yang gagal saja untuk DAG tertentu, gunakan juga argumen --rerun_failed_tasks.

Ganti:

  • ENVIRONMENT_NAME dengan nama lingkungan.
  • LOCATION dengan region tempat lingkungan berada.
  • START_DATE dengan nilai untuk parameter DAG start_date, dalam format YYYY-MM-DD.
  • END_DATE dengan nilai untuk parameter DAG end_date, dalam format YYYY-MM-DD.
  • DAG_NAME dengan nama DAG.

Operasi pengisian ulang terkadang dapat menghasilkan situasi deadlock saat pengisian ulang tidak dapat dilakukan karena ada kunci pada tugas. Contoh:

2022-11-08 21:24:18.198 CET DAG ID Task ID Run ID Try number
2022-11-08 21:24:18.201 CET -------- --------- -------- ------------
2022-11-08 21:24:18.202 CET 2022-11-08 21:24:18.203 CET These tasks are deadlocked:
2022-11-08 21:24:18.203 CET DAG ID Task ID Run ID Try number
2022-11-08 21:24:18.204 CET ----------------------- ----------- ----------------------------------- ------------
2022-11-08 21:24:18.204 CET <DAG name> <Task name> backfill__2022-10-27T00:00:00+00:00 1
2022-11-08 21:24:19.249 CET Command exited with return code 1
...
2022-11-08 21:24:19.348 CET Failed to execute job 627927 for task backfill

Dalam beberapa kasus, Anda dapat menggunakan solusi berikut untuk mengatasi deadlock:

  • Nonaktifkan mini-scheduler dengan mengganti the [core]schedule_after_task_execution ke False.

  • Jalankan pengisian ulang untuk rentang tanggal yang lebih sempit. Misalnya, tetapkan START_DATE dan END_DATE untuk menentukan periode hanya 1 hari.

Langkah berikutnya