Memecahkan masalah penjadwal Airflow

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Halaman ini memberikan langkah-langkah pemecahan masalah dan informasi untuk masalah umum pada penjadwal Airflow dan pemroses DAG.

Mengidentifikasi sumber masalah

Untuk mulai memecahkan 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 parsing dan waktu eksekusi, baca artikel Perbedaan antara waktu parsing DAG dan waktu eksekusi DAG.

Memeriksa masalah pemrosesan DAG

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

Memantau tugas yang sedang berjalan dan dalam antrean

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

  1. Di Google Cloud console, buka halaman Environments.

    Buka Lingkungan

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

  3. Buka tab Monitoring.

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

Memecahkan masalah saat waktu penguraian DAG

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

Penguraian dan penjadwalan DAG di Cloud Composer 1 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 Cloud Composer 1, penjadwal berjalan di node cluster bersama dengan komponen Cloud Composer lainnya. Oleh karena itu, beban setiap node cluster mungkin lebih tinggi atau lebih rendah dibandingkan dengan node lainnya. Performa penjadwal (penguraian dan penjadwalan DAG) dapat bervariasi, bergantung pada node tempat penjadwal berjalan. Selain itu, node individual tempat penjadwal berjalan dapat berubah sebagai akibat dari operasi upgrade atau pemeliharaan. Batasan ini telah diatasi di Cloud Composer 2, tempat Anda dapat mengalokasikan resource CPU dan memori ke penjadwal dan performa penjadwal tidak bergantung pada beban node cluster.

Jumlah dan distribusi waktu tugas

Airflow dapat mengalami masalah saat menjadwalkan sejumlah besar DAG atau tugas secara bersamaan. Untuk menghindari masalah penjadwalan, Anda dapat:

  • Sesuaikan DAG Anda untuk menggunakan lebih sedikit tugas yang lebih digabungkan.
  • Sesuaikan interval jadwal DAG untuk mendistribusikan operasi DAG secara lebih merata dari waktu ke waktu.

Konfigurasi penskalaan Airflow

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

  • Konkurensi pekerja

    Parameter [celery]worker_concurrency mengontrol jumlah maksimum tugas yang dapat dijalankan oleh pekerja Airflow secara bersamaan. Jika Anda mengalikan nilai parameter ini dengan jumlah worker Airflow di lingkungan Cloud Composer, Anda akan mendapatkan jumlah maksimum tugas yang dapat dieksekusi pada saat 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. Penjadwal tidak membuat lebih banyak proses DAG jika mencapai batas ini.

    Jika parameter ini disetel dengan tidak benar, Anda mungkin mengalami masalah saat penjadwal membatasi eksekusi DAG karena tidak dapat membuat lebih banyak instance DAG yang berjalan dalam 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 disetel dengan tidak benar, Anda mungkin mengalami masalah saat eksekusi satu instance DAG berjalan lambat karena hanya ada sejumlah kecil tugas DAG yang dapat dieksekusi 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 parameter max_active_tis_per_dag dan max_active_tis_per_dagrun di tingkat tugas untuk mengontrol jumlah instance dengan ID tugas tertentu yang diizinkan untuk berjalan per DAG dan per eksekusi DAG.

  • Paralelisme dan ukuran kumpulan

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

    Ini adalah parameter global untuk seluruh penyiapan Airflow.

    Tugas diantrekan dan dieksekusi dalam pool. Lingkungan Cloud Composer hanya menggunakan satu pool. Ukuran pool ini mengontrol jumlah tugas yang dapat diantrekan oleh penjadwal untuk dieksekusi pada waktu tertentu. Jika ukuran pool terlalu kecil, penjadwal 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 pekerja Airflow, belum terpenuhi.

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

    Biasanya, [core]parallelism ditetapkan sebagai produk dari jumlah maksimum pekerja 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 sedang 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 dilakukan 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.

  • Menetapkan start_date statis untuk DAG. Sebagai opsi, Anda dapat menggunakan catchup=False untuk menonaktifkan DAG agar tidak 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 penjadwal 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 Jadwal Bawaan.

Resource cluster terbatas

Anda mungkin mengalami masalah performa jika cluster GKE di 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 tinggi dan migrasikan DAG Anda ke lingkungan tersebut.
  • Buat lebih banyak lingkungan Cloud Composer 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 terhadap error, opsi ini paling tidak direkomendasikan.
  • Upgrade jenis mesin instance Cloud SQL yang menjalankan database Airflow di lingkungan Anda, misalnya menggunakan perintah gcloud composer environments update. Performa database Airflow yang rendah mungkin menjadi alasan mengapa penjadwal berjalan lambat.

Menghindari penjadwalan tugas selama masa pemeliharaan

Anda dapat menentukan masa pemeliharaan untuk lingkungan sehingga pemeliharaan lingkungan terjadi di luar waktu saat Anda menjalankan DAG. Anda tetap dapat menjalankan DAG selama masa pemeliharaan, asalkan 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, maka agar tugas berhasil, semua tugas yang langsung downstream dari tugas ini juga harus berhasil. Artinya, eksekusi tugas yang termasuk dalam DAG run tertentu dapat diperlambat oleh eksekusi tugas dari DAG run sebelumnya. Baca selengkapnya di dokumentasi Airflow.

Tugas yang terlalu lama dalam antrean akan dibatalkan dan dijadwalkan ulang

Jika tugas Airflow tetap berada dalam antrean terlalu lama, penjadwal akan menjadwalkannya ulang untuk dieksekusi setelah jangka waktu yang ditetapkan dalam opsi konfigurasi Airflow [scheduler]task_queued_timeout telah berlalu. Nilai defaultnya adalah 2400. Pada versi Airflow sebelum 2.3.1, tugas juga ditandai sebagai gagal dan dicoba lagi jika memenuhi syarat untuk percobaan ulang.

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

Secara umum, perilaku ini sudah diperkirakan dan instance berikutnya dari tugas terjadwal dimaksudkan untuk dijalankan sesuai jadwal. Jika Anda mengamati banyak kasus seperti itu di lingkungan Cloud Composer, hal ini mungkin berarti tidak ada cukup banyak worker Airflow di lingkungan Anda untuk memproses semua tugas terjadwal.

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

Pendekatan Cloud Composer terhadap parameter min_file_process_interval

Cloud Composer mengubah cara [scheduler]min_file_process_interval digunakan oleh penjadwal Airflow.

Aliran udara 1

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

Airflow 2

Pada Cloud Composer versi yang lebih lama dari 1.19.9, [scheduler]min_file_process_interval diabaikan.

Cloud Composer versi yang lebih baru dari 1.19.9:

Scheduler Airflow dimulai ulang setelah semua DAG dijadwalkan beberapa kali dan parameter [scheduler]num_runs mengontrol berapa kali hal itu dilakukan oleh scheduler. Saat penjadwal mencapai loop penjadwalan [scheduler]num_runs, penjadwal akan dimulai ulang. Scheduler adalah komponen stateless dan mulai ulang tersebut 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 seberapa sering penguraian DAG terjadi, tetapi parameter ini tidak boleh lebih lama dari waktu yang diperlukan untuk penjadwal melakukan loop [scheduler]num_runs saat menjadwalkan DAG Anda.

Menandai tugas sebagai gagal setelah mencapai dagrun_timeout

Penjadwal menandai tugas yang belum selesai (berjalan, terjadwal, dan dalam antrean) sebagai gagal jika eksekusi DAG tidak selesai dalam dagrun_timeout (parameter DAG).

Solusi:

Gejala database Airflow mengalami beban berat

Terkadang di log penjadwal 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 dalam log pekerja 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 dalam waktu yang sama, baik oleh penjadwal maupun oleh komponen Airflow lainnya seperti pekerja, pemicu, dan server web.

Kemungkinan solusi:

Server web menampilkan peringatan 'The scheduler does not appear to be running' (Penjadwal tampaknya tidak berjalan)

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

Terkadang, jika penjadwal mengalami beban berat, penjadwal mungkin tidak dapat melaporkan detak jantungnya 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 penjadwal.

  • Optimalkan DAG agar parsing dan penjadwalannya lebih cepat dan tidak menggunakan terlalu banyak resource penjadwal.

  • 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 Airflow CLI 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 hanya tugas yang gagal untuk DAG tertentu, gunakan juga argumen --rerun-failed-tasks.

Aliran udara 1 

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

Untuk menjalankan ulang hanya tugas yang gagal 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 menyebabkan situasi kebuntuan 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 sementara berikut untuk mengatasi kebuntuan:

  • Nonaktifkan penjadwal mini dengan mengganti [core]schedule_after_task_execution menjadi 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