Memecahkan masalah penjadwal Airflow

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Halaman ini memberikan langkah-langkah pemecahan masalah dan informasi untuk terkait dengan penjadwal Airflow.

Mengidentifikasi sumber masalah

Untuk memulai pemecahan 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, berarti Prosesor DAG, yang dijalankan oleh penjadwal, mungkin tidak mengurai semua DAG. Hal ini dapat menyebabkan banyak masalah yang memiliki gejala berikut.

Gejala:

  • Jika Pemroses 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.

  • Terjadi 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 scheduler dimulai ulang.

  • Tugas Airflow yang dijadwalkan untuk dieksekusi akan dibatalkan dan DAG dijalankan untuk DAG yang gagal diurai 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:

  • Memperbaiki atau menghapus DAG yang menyebabkan masalah pada prosesor DAG.

Memeriksa waktu penguraian DAG

Untuk memastikan 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 Cloud Composer Monitoring:

  1. Di Konsol Google Cloud, buka halaman Environments.

    Buka Lingkungan

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

  3. Di tab Monitoring, tinjau Total parse time for semua DAG file di bagian DAG dijalankan dan mengidentifikasi kemungkinan masalah.

    Bagian operasi DAG di tab Pemantauan Composer menampilkan metrik respons 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. Pada 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 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 nilai yang tercantum dalam tabel. Nilai yang besar mungkin menunjukkan bahwa salah satu DAG Anda tidak diterapkan secara optimal. Dari tabel output, Anda dapat mengidentifikasi DAG yang memiliki waktu penguraian yang lama.

Memantau tugas yang berjalan dan dalam antrean

Untuk memeriksa apakah Anda memiliki tugas yang terjebak dalam antrean, ikuti langkah-langkah berikut.

  1. Di Konsol Google Cloud, buka halaman Environments.

    Buka Lingkungan

  2. Pada daftar lingkungan, klik nama lingkungan Anda. Halaman Detail lingkungan akan terbuka.

  3. Buka tab Monitoring.

  4. Di tab Monitoring, tinjau diagram Airflow tasks di bagian DAG berjalan dan mengidentifikasi kemungkinan masalah. Tugas Airflow adalah tugas yang berada dalam status antrean di Airflow, tugas tersebut dapat Antrean broker Celery atau Kubernetes Executor. Tugas dalam antrean seledri adalah tugas yang dimasukkan ke dalam antrean broker Celery.

Memecahkan masalah pada waktu penguraian DAG

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

Jumlah rangkaian pesan terbatas

Mengizinkan pengelola prosesor DAG (bagian dari penjadwal yang memproses file DAG) agar hanya menggunakan thread dalam jumlah terbatas dapat berdampak waktu penguraian DAG Anda.

Untuk mengatasi masalah tersebut, 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
    dalam 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
    dalam mesin node pekerja.

Distribusi jumlah dan waktu tugas

Airflow dikenal karena bermasalah dengan penjadwalan tugas klasifikasi. Dalam situasi seperti itu, Anda harus memilih untuk menerima lebih sedikit tugas yang terkonsolidasi.

Menjadwalkan sejumlah besar DAG atau tugas secara bersamaan juga dapat menjadi sumber masalah yang mungkin terjadi. 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 ini menjelaskan gejala dan kemungkinan perbaikan untuk beberapa masalah terkait tugas yang berjalan dan dalam antrean.

Task queue terlalu panjang

Dalam beberapa kasus, task queue mungkin terlalu panjang untuk penjadwal. Untuk informasi tentang cara mengoptimalkan parameter pekerja dan seledri, baca tentang menskalakan lingkungan Cloud Composer bersama bisnis Anda.

Menggunakan fitur TimeTable penjadwal Airflow

Dimulai dengan 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 kinerja jika cluster GKE lingkungan Anda terlalu kecil untuk menangani semua DAG dan tugas. Dalam hal ini, coba salah satu solusi berikut:

  • Menciptakan lingkungan baru dengan jenis mesin yang memberikan performa lebih baik dan memigrasikan DAG Anda ke sana.
  • Membuat lebih banyak lingkungan Cloud Composer dan membagi DAG di antara mereka.
  • Mengubah jenis mesin untuk node GKE, seperti yang dijelaskan di Mengupgrade jenis mesin untuk node GKE. Karena prosedur ini rentan mengalami error, prosedur ini adalah opsi yang paling tidak direkomendasikan.
  • Mengupgrade jenis mesin instance Cloud SQL yang menjalankan Airflow di lingkungan Anda, misalnya menggunakan database Perintah gcloud composer environments update. Performa yang rendah dari database Airflow mungkin menjadi penyebab penjadwalnya lambat.

Hindari penjadwalan tugas selama masa pemeliharaan

Anda dapat menentukan masa pemeliharaan khusus untuk lingkungan fleksibel App Engine. Selama jangka waktu ini, peristiwa pemeliharaan untuk Cloud SQL dan GKE berlangsung.

Membuat penjadwal Airflow mengabaikan file yang tidak perlu

Anda dapat meningkatkan performa scheduler Airflow dengan melewati langkah yang tidak perlu di folder DAG. Scheduler Airflow mengabaikan file dan folder yang ditentukan dalam file .airflowignore.

Agar scheduler Airflow mengabaikan file yang tidak diperlukan:

  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.

Untuk informasi selengkapnya tentang format file .airflowignore, lihat Dokumentasi Airflow.

Proses penjadwal Airflow menjeda DAG

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

Scheduler Airflow akan terus mengurai DAG yang dijeda. Jika Anda benar-benar ingin tingkatkan performa penjadwal Airflow, gunakan .airflowignore atau hapus yang dijeda DAG 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 berada di downstream dari tugas juga harus berhasil. Ini berarti bahwa pelaksanaan tugas-tugas yang dimiliki oleh proses DAG tertentu mungkin diperlambat oleh eksekusi tugas dari Operasi DAG. Baca selengkapnya di dokumentasi Airflow.

Tugas dalam antrean terlalu lama akan dibatalkan dan dijadwalkan ulang

Jika tugas Airflow disimpan dalam antrean terlalu lama, scheduler akan menjadwal ulang lagi untuk eksekusi (di versi Airflow lebih awal dari 2.3.1, tugas juga ditandai sebagai gagal dan dicoba lagi jika memenuhi syarat untuk dicoba ulang).

Salah satu cara untuk mengamati gejala situasi ini adalah dengan melihat bagan dengan jumlah tugas yang diantrekan (Tab "Monitoring" di UI Cloud Composer) dan jika lonjakan dalam diagram ini tidak turun dalam waktu sekitar dua jam, tugas-tugas kemungkinan besar akan dijadwalkan ulang (tanpa log) diikuti oleh "Tugas yang diterapkan masih tertunda ..." entri log di log penjadwal. Dalam kasus tersebut, Anda mungkin melihat pesan "File log tidak ditemukan..." pesan di log tugas Airflow karena tugas tidak dijalankan.

Secara umum, perilaku ini diharapkan dan instance berikutnya dari tugas ini dimaksudkan untuk dieksekusi sesuai jadwal. Jika Anda mengamati bahwa kasus semacam itu di lingkungan Cloud Composer, mungkin berarti bahwa tidak ada cukup pekerja Airflow di lingkungan Anda untuk memproses semua tugas-tugas yang dijadwalkan.

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

Tugas yang sudah tidak berlaku secara sporadis dapat memblokir eksekusi DAG tertentu

Dalam kasus biasa, scheduler Airflow harus dapat menangani situasi di mana ada tugas-tugas basi dalam antrean dan untuk beberapa alasan itu mungkin untuk mengeksekusinya dengan benar (mis. DAG tempat tugas lama berada dihapus).

Jika tugas-tugas basi ini tidak dibersihkan oleh {i>scheduler<i}, maka Anda mungkin perlu menghapusnya secara manual. Anda dapat melakukannya, misalnya, di UI Airflow - Anda dapat buka (Menu &gt; Browser &gt; Instance Tugas), temukan tugas dalam antrean milik DAG yang sudah tidak berlaku, lalu hapus.

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

Pendekatan Cloud Composer terhadap parameter [scheduler]min_file_process_interval

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

Aliran udara 1

Pada Cloud Composer yang menggunakan Airflow 1, pengguna dapat menyetel nilai dari [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.

Aliran udara 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

  • Versi Cloud Composer yang lebih lama dari 1.19.9 dan 2.0.26

    Dalam versi ini, [scheduler]min_file_process_interval akan diabaikan.

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

    Scheduler Airflow dimulai ulang setelah beberapa kali semua DAG dijadwalkan dan parameter [scheduler]num_runs mengontrol frekuensi tindakan ini dilakukan oleh penjadwal. Waktu penjadwal mencapai [scheduler]num_runs loop penjadwalan, {i>restart<i} - Scheduler adalah komponen stateless dan proses {i>restart<i} tersebut mekanisme {i>auto-healing <i}untuk setiap masalah yang mungkin dialami Penjadwal. Bila tidak ditentukan, nilai default nilai [scheduler]num_runs diterapkan, yaitu 5.000.

    [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 guna melakukan [scheduler]num_runs loop saat menjadwalkan DAG.

Konfigurasi Penskalaan Aliran Udara

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

  • Serentak 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 Google Cloud Composer, Anda akan mendapatkan jumlah maksimum tugas yang dapat dieksekusi pada waktu tertentu di lingkungan Anda. Ini nomor 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 32, 12 * worker_CPU, dan 8 * worker_memory.

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

  • Operasi DAG Aktif Maksimum

    Kontrol opsi konfigurasi Airflow [core]max_active_runs_per_dag jumlah maksimum operasi DAG aktif per DAG. Penjadwal tidak membuat lebih banyak operasi 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 DAG menjalankan instance pada waktu tertentu.

  • Tugas Aktif Maks Per DAG

    Opsi konfigurasi Airflow [core]max_active_tasks_per_dag mengontrol jumlah maksimum instance tugas yang dapat berjalan serentak di setiap DAG. Parameter ini adalah parameter tingkat DAG.

    Jika parameter ini tidak disetel dengan benar, Anda mungkin mengalami masalah di mana eksekusi satu instance DAG lambat karena hanya ada tugas DAG dalam jumlah terbatas 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 penjadwal Airflow di antrean Executor setelah dependensi untuk tugas-tugas ini terpenuhi.

    Ini adalah parameter global untuk seluruh penyiapan Airflow.

    Tugas diantrekan dan dieksekusi dalam kumpulan. Cloud Composer lingkungan menggunakan hanya satu kumpulan. Ukuran kolam ini mengontrol berapa jumlah tugas dapat diantrekan oleh penjadwal untuk dieksekusi dalam waktu tertentu. Jika ukuran kumpulan terlalu kecil, maka penjadwal tidak dapat mengantrekan tugas untuk eksekusi meskipun dengan nilai minimum, yang ditentukan oleh [core]parallelism opsi konfigurasi dan dengan opsi konfigurasi [celery]worker_concurrency dikalikan dengan jumlah worker Airflow, belum terpenuhi.

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

    Biasanya, [core]parallelism ditetapkan sebagai perkalian dari jumlah maksimum pekerja dan [celery]worker_concurrency.

DAG tidak dijadwalkan oleh penjadwal karena waktu tunggu prosesor 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, terjadwal, dan dalam antrean) sebagai gagal jika operasi DAG tidak selesai dalam dagrun_timeout (parameter DAG).

Solusi:

Gejala Database Airflow sedang di bawah tekanan beban

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 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

Kesalahan atau peringatan tersebut mungkin merupakan gejala dari {i>database<i} Airflow yang kewalahan dengan jumlah koneksi terbuka atau jumlah kueri dijalankan secara bersamaan, baik oleh penjadwal atau komponen Airflow lainnya seperti worker, pemicu, dan server web.

Solusi yang memungkinkan:

Server web menampilkan 'Penjadwal tampaknya tidak berjalan' peringatan

Scheduler melaporkan detak jantungnya secara berkala ke Airflow di skrip untuk menyiapkan database. Berdasarkan informasi ini, server web Airflow menentukan apakah penjadwal aktif.

Terkadang, jika penjadwal memiliki beban yang berat, maka sistem itu mungkin tidak dapat melaporkan detak jantungnya setiap [scheduler]scheduler-heartbeat-sec.

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

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

Solusi yang memungkinkan:

  • Meningkatkan resource CPU dan memori untuk penjadwal.

  • Optimalkan DAG Anda agar penguraian dan penjadwalannya jika lebih cepat dan tidak cepat menghabiskan terlalu banyak sumber daya penjadwal.

  • Hindari penggunaan variabel global di DAG Airflow: Variabel lingkungan Cloud Composer dan Variabel Airflow.

  • Tingkatkan nilai [scheduler]scheduler-health-check-threshold sehingga server web menunggu lebih lama sebelum melaporkan ketidaktersediaan {i>scheduler<i} (penjadwal).

Solusi untuk masalah yang ditemukan selama mengisi 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 menjalankan kembali tugas yang gagal saja untuk DAG tertentu, gunakan juga Argumen --rerun_failed_tasks.

Aliran udara 2

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

Untuk menjalankan kembali 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, di dalam format YYYY-MM-DD.
  • END_DATE dengan nilai untuk parameter DAG end_date, di dalam format YYYY-MM-DD.
  • DAG_NAME dengan nama DAG.

Operasi pengisian ulang terkadang menghasilkan situasi deadlock di mana isi 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 ke False.

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

Langkah selanjutnya