Memecahkan masalah penjadwal Airflow

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

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

Mengidentifikasi sumber masalah

Untuk mulai memecahkan masalah, identifikasi apakah masalah terjadi pada waktu penguraian DAG atau saat memproses tugas pada waktu eksekusi. Untuk informasi selengkapnya tentang waktu penguraian dan waktu eksekusi, baca Perbedaan antara waktu penguraian DAG dan waktu eksekusi DAG.

Memeriksa log Prosesor DAG

Jika Anda memiliki DAG yang kompleks, Pemroses DAG, yang dijalankan oleh penjadwal, mungkin tidak mengurai semua DAG Anda. Hal ini dapat menyebabkan banyak masalah yang memiliki gejala berikut.

Gejala:

  • Jika Prosesor DAG mengalami masalah saat mengurai DAG, hal ini dapat menyebabkan kombinasi masalah yang tercantum di bawah. Jika DAG dibuat secara dinamis, masalah ini mungkin lebih berdampak dibandingkan dengan DAG statis.

  • DAG tidak terlihat di UI Airflow dan UI DAG.

  • DAG tidak dijadwalkan untuk dieksekusi.

  • Ada error dalam log pemroses DAG, misalnya:

    dag-processor-manager [2023-04-21 21:10:44,510] {manager.py:1144} ERROR -
Processor for /home/airflow/gcs/dags/dag-example.py with PID 68311 started
at 2023-04-21T21:09:53.772793+00:00 has timed out, killing it.

    atau

    dag-processor-manager [2023-04-26 06:18:34,860] {manager.py:948} ERROR -
Processor for /home/airflow/gcs/dags/dag-example.py exited with return
code 1.

  • Penjadwal Airflow mengalami masalah yang menyebabkan penjadwal dimulai ulang.

  • Tugas Airflow yang dijadwalkan untuk dieksekusi dibatalkan dan DAG berjalan untuk DAG yang gagal diuraikan mungkin ditandai sebagai failed. Contoh:

    airflow-scheduler Failed to get task '<TaskInstance: dag-example.task1--1
manual__2023-04-17T10:02:03.137439+00:00 [removed]>' for dag
'dag-example'. Marking it as removed.

Solusi:

  • Meningkatkan parameter yang terkait dengan penguraian DAG:

    • Tingkatkan dagbag-import-timeout menjadi minimal 120 detik (atau lebih, jika diperlukan).

    • Tingkatkan dag-file-processor-timeout menjadi minimal 180 detik (atau lebih, jika diperlukan). Nilai ini harus lebih tinggi dari dagbag-import-timeout.

  • Perbaiki atau hapus DAG yang menyebabkan masalah pada pemroses DAG.

Memeriksa waktu penguraian DAG

Untuk memverifikasi apakah masalah terjadi pada waktu penguraian DAG, ikuti langkah-langkah berikut.

Konsol

Di konsol Google Cloud, Anda dapat menggunakan halaman Monitoring dan tab Logs untuk memeriksa waktu penguraian DAG.

Periksa waktu penguraian DAG dengan halaman Pemantauan Cloud Composer:

  1. Di konsol Google Cloud, buka halaman Environments.

    Buka Lingkungan

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

  3. Di tab Pemantauan, tinjau diagram Total waktu penguraian untuk semua file DAG di bagian DAG berjalan dan identifikasi kemungkinan masalah.

    Bagian DAG berjalan di tab Composer Monitoring menampilkan metrik kondisi untuk DAG di lingkungan Anda

Periksa waktu penguraian DAG dengan tab Cloud Composer Logs:

  1. Di konsol Google Cloud, buka halaman Environments.

    Buka Lingkungan

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

  3. Buka tab Logs, dan dari hierarki navigasi All logs, pilih bagian DAG processor manager.

  4. Tinjau log dag-processor-manager dan identifikasi kemungkinan masalah.

    Log pemroses DAG akan menampilkan waktu penguraian DAG

gcloud

Gunakan perintah dags report untuk melihat waktu penguraian untuk semua DAG Anda.

gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags report

Ganti:

  • ENVIRONMENT_NAME dengan nama lingkungan.
  • LOCATION dengan region tempat lingkungan tersebut berada.

Output perintah terlihat mirip dengan berikut ini:

Executing within the following Kubernetes cluster namespace: composer-2-0-31-airflow-2-3-3
file                  | duration       | dag_num | task_num | dags
======================+================+=========+==========+===================
/manydagsbig.py       | 0:00:00.038334 | 2       | 10       | serial-0,serial-0
/airflow_monitoring.py| 0:00:00.001620 | 1       | 1        | airflow_monitoring

Cari nilai duration untuk setiap dag yang tercantum dalam tabel. Nilai yang besar mungkin menunjukkan bahwa salah satu DAG Anda tidak diterapkan dengan cara yang optimal. Dari tabel output, Anda dapat mengidentifikasi DAG mana yang memiliki waktu penguraian yang lama.

Memantau tugas yang berjalan dan dalam antrean

Untuk memeriksa apakah Anda memiliki tugas yang macet di antrean, ikuti langkah-langkah berikut.

  1. Di konsol Google Cloud, buka halaman Environments.

    Buka Lingkungan

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

  3. Buka tab Monitoring.

  4. Di tab Pemantauan, tinjau diagram Tugas Airflow di bagian DAG berjalan dan identifikasi kemungkinan masalah. Tugas Airflow adalah tugas yang berada dalam status antrean di Airflow, yang 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 pada waktu penguraian DAG

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

Jumlah thread terbatas

Mengizinkan pengelola pemroses DAG (bagian dari penjadwal yang memproses file DAG) untuk hanya menggunakan sejumlah thread terbatas dapat memengaruhi waktu penguraian DAG Anda.

Untuk mengatasi masalah ini, ganti opsi konfigurasi Airflow berikut:

  • Untuk Airflow 1.10.12 dan versi sebelumnya, ganti parameter max_threads:

    Bagian Kunci Nilai Catatan
    scheduler max_threads NUMBER_OF_CORES_IN_MACHINE - 1 Ganti NUMBER_OF_CORES_IN_MACHINE dengan jumlah core
    di mesin node pekerja.

  • Untuk Airflow 1.10.14 dan versi yang lebih baru, ganti parameter parsing_processes:

    Bagian Kunci Nilai Catatan
    scheduler parsing_processes NUMBER_OF_CORES_IN_MACHINE - 1 Ganti NUMBER_OF_CORES_IN_MACHINE dengan jumlah core
    di mesin node pekerja.

Distribusi jumlah dan waktu tugas

Airflow diketahui memiliki masalah dalam menjadwalkan sejumlah besar tugas kecil. Dalam situasi seperti itu, Anda harus memilih tugas yang lebih sedikit dan lebih dikonsolidasikan.

Menjadwalkan sejumlah besar DAG atau tugas secara bersamaan juga dapat menjadi sumber masalah. Untuk menghindari masalah ini, distribusikan tugas Anda secara lebih merata dari waktu ke waktu.

Memecahkan masalah terkait tugas yang berjalan dan dalam antrean

Bagian berikut menjelaskan gejala dan kemungkinan perbaikan untuk beberapa masalah umum tentang tugas yang berjalan dan diantrekan.

Task queue terlalu panjang

Dalam beberapa kasus, antrean tugas mungkin terlalu panjang untuk penjadwal. Untuk informasi tentang cara mengoptimalkan parameter pekerja dan celery, baca artikel tentang menskalakan lingkungan Cloud Composer bersama bisnis Anda.

Menggunakan fitur TimeTable dari penjadwal Airflow

Mulai Airflow 2.2, Anda dapat menentukan tabel waktu untuk DAG menggunakan fitur baru yang disebut TimeTable.

Anda dapat menentukan tabel waktu menggunakan salah satu metode berikut:

Resource cluster terbatas

Bagian ini hanya berlaku untuk Cloud Composer 1.

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

  • Buat lingkungan baru dengan jenis mesin yang memberikan lebih banyak performa dan migrasikan DAG Anda ke sana.
  • Buat lebih banyak lingkungan Cloud Composer dan bagi DAG-nya 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, opsi ini adalah opsi yang 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 penjadwal lambat.

Menghindari penjadwalan tugas selama periode pemeliharaan

Anda dapat menentukan masa pemeliharaan tertentu untuk lingkungan Anda. Selama jangka waktu ini, peristiwa pemeliharaan untuk Cloud SQL dan GKE akan berlangsung.

Membuat penjadwal Airflow mengabaikan file yang tidak diperlukan

Anda dapat meningkatkan performa penjadwal Airflow dengan melewati file yang tidak diperlukan di folder DAG. Penjadwal Airflow mengabaikan file dan folder yang ditentukan dalam file .airflowignore.

Agar penjadwal Airflow mengabaikan file yang tidak perlu:

  1. Buat file .airflowignore.
  2. Dalam file ini, cantumkan file dan folder yang harus diabaikan.
  3. Upload file ini ke folder /dags di bucket lingkungan Anda.

Untuk mengetahui informasi selengkapnya tentang format file .airflowignore, lihat dokumentasi Airflow.

Airflow scheduler memproses DAG yang dijeda

Pengguna Airflow menjeda DAG untuk menghindari eksekusinya. Hal ini menghemat siklus pemrosesan pekerja Airflow.

Penjadwal Airflow akan terus mengurai DAG yang dijeda. Jika Anda benar-benar ingin meningkatkan performa penjadwal Airflow, gunakan .airflowignore atau hapus DAG yang dijeda dari folder DAG.

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 eksekusi DAG tertentu mungkin akan diperlambat oleh eksekusi tugas dari eksekusi DAG sebelumnya. Baca selengkapnya di dokumentasi Airflow.

Tugas yang berada dalam antrean terlalu lama akan dibatalkan dan dijadwalkan ulang

Jika tugas Airflow disimpan dalam antrean terlalu lama, penjadwal akan menjadwalkannya ulang untuk dieksekusi (di versi Airflow yang lebih lama dari 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 yang berisi jumlah tugas dalam antrean (tab "Monitoring" di UI Cloud Composer) dan jika lonjakan dalam diagram ini tidak turun dalam waktu sekitar dua jam, maka tugas kemungkinan besar akan dijadwalkan ulang (tanpa log) diikuti dengan entri log "Tugas yang diadopsi masih tertunda ..." di log penjadwal. Dalam kasus tersebut, Anda mungkin melihat pesan "Log file is not found..." di log tugas Airflow karena tugas tidak dieksekusi.

Secara umum, perilaku ini sudah diperkirakan dan instance tugas terjadwal berikutnya dimaksudkan untuk dijalankan sesuai jadwal. Jika Anda mengamati banyak kasus seperti itu di lingkungan Cloud Composer, hal ini mungkin berarti tidak ada cukup pekerja 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 pekerja atau worker_concurrency. Anda juga dapat menyesuaikan paralelisme atau kumpulan untuk mencegah tugas antrean melebihi kapasitas yang Anda miliki.

Terkadang, tugas yang sudah tidak berlaku dapat memblokir eksekusi DAG tertentu

Dalam kasus biasa, penjadwal Airflow harus dapat menangani situasi saat ada tugas yang sudah tidak berlaku dalam antrean dan karena alasan tertentu, tidak mungkin untuk menjalankannya dengan benar (misalnya, DAG tempat tugas yang sudah tidak berlaku dihapus).

Jika tugas yang sudah tidak berlaku ini tidak dihapus oleh penjadwal, Anda mungkin perlu menghapusnya secara manual. Anda dapat melakukannya, misalnya, di UI Airflow - Anda dapat membuka (Menu > Browser > Task Instances), menemukan tugas yang diantrekan yang termasuk dalam DAG yang sudah tidak berlaku dan menghapusnya.

Untuk mengatasi masalah ini, upgrade lingkungan Anda ke Cloud Composer versi 2.1.12 atau yang lebih baru.

Pendekatan Cloud Composer untuk parameter [scheduler]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 akan memberikan hasil yang sama seperti jika [scheduler]min_file_process_interval ditetapkan ke 600 detik.

Airflow 2

Di Airflow 2, [scheduler]min_file_process_interval hanya dapat digunakan dengan versi 1.19.9 dan 2.0.26 atau yang lebih baru

  • Cloud Composer versi yang lebih lama dari 1.19.9 dan 2.0.26

    Dalam versi ini, [scheduler]min_file_process_interval diabaikan.

  • Cloud Composer versi 1.19.9 atau 2.0.26, atau versi yang lebih baru

    Penjadwal Airflow dimulai ulang setelah sejumlah waktu tertentu semua DAG dijadwalkan dan parameter [scheduler]num_runs mengontrol frekuensi penjadwalan oleh penjadwal. Saat penjadwal mencapai loop penjadwalan [scheduler]num_runs, penjadwal akan dimulai ulang - Penjadwal adalah komponen stateless dan mulai ulang tersebut adalah mekanisme perbaikan otomatis untuk masalah apa pun yang mungkin dialami Penjadwal. Jika tidak ditentukan, nilai default [scheduler]num_runs akan diterapkan, yaitu 5.000.

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

Menskalakan konfigurasi Airflow

Airflow menyediakan opsi konfigurasi Airflow yang mengontrol jumlah tugas dan DAG yang dapat dieksekusi Airflow secara bersamaan. Untuk menetapkan opsi konfigurasi ini, ganti nilainya untuk lingkungan Anda.

  • Konkurensi Pekerja

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

    Di lingkungan Cloud Composer 2, nilai default [celery]worker_concurrency dihitung secara otomatis

    • Untuk versi Airflow: 2.3.3 dan yang lebih baru, [celery]worker_concurrency ditetapkan ke nilai minimum dari 32, 12 * worker_CPU, dan 8 * worker_memory.

    • Untuk versi Airflow: 2.2.5 atau yang lebih lama, [celery]worker_concurrency ditetapkan ke 12 * jumlah CPU pekerja.

  • Pemrosesan DAG Aktif Maksimal

    Opsi konfigurasi Airflow [core]max_active_runs_per_dag mengontrol jumlah maksimum DAG aktif yang berjalan per DAG. Penjadwal tidak akan membuat lebih banyak DAG berjalan jika mencapai batas ini.

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

  • 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. Ini adalah parameter tingkat DAG.

    Jika parameter ini ditetapkan dengan tidak benar, Anda mungkin mengalami masalah saat eksekusi satu instance DAG lambat karena hanya ada sejumlah terbatas tugas DAG yang dapat dieksekusi pada waktu tertentu.

    Solusi: tingkatkan [core]max_active_tasks_per_dag.

  • Paralelisme dan ukuran kumpulan

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

    Ini adalah parameter global untuk seluruh penyiapan Airflow.

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

DAG tidak dijadwalkan oleh penjadwal karena waktu tunggu pemroses DAG habis

Untuk mengetahui informasi selengkapnya tentang masalah ini, lihat Memecahkan Masalah DAG.

Menandai tugas sebagai gagal setelah mencapai dagrun_timeout

Penjadwal 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 berada di bawah tekanan beban

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 dibanjiri dengan jumlah koneksi terbuka atau jumlah kueri yang dijalankan secara bersamaan, baik oleh penjadwal atau komponen Airflow lainnya seperti pekerja, pemicu, dan server web.

Kemungkinan solusi:

Server web menampilkan peringatan 'Penjadwal tampaknya tidak berjalan'

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

Terkadang, jika scheduler mengalami beban berat, scheduler mungkin tidak dapat melaporkan heartbeat 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:

Solusi untuk masalah yang dialami selama pengisian ulang DAG

Terkadang, Anda mungkin ingin menjalankan ulang DAG yang sudah dieksekusi. Anda dapat melakukannya dengan alat command line Airflow dengan cara berikut:

Aliran udara 1

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

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

Airflow 2

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

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

Ganti:

  • ENVIRONMENT_NAME dengan nama lingkungan.
  • LOCATION dengan region tempat lingkungan tersebut 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 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 selanjutnya