Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1
Tutorial ini memandu Anda mendiagnosis dan memecahkan masalah penjadwalan tugas dan masalah parsing yang menyebabkan malfungsi penjadwal, error parsing dan latensi, serta kegagalan tugas.
Pengantar
Scheduler Airflow terutama dipengaruhi oleh dua faktor: penjadwalan tugas dan penguraian DAG. Masalah pada salah satu faktor tersebut dapat berdampak negatif pada kesehatan dan performa lingkungan.
Terkadang, terlalu banyak tugas yang dijadwalkan secara bersamaan. Dalam situasi ini, antrean akan penuh, dan tugas tetap dalam status "dijadwalkan" atau dijadwalkan ulang setelah diantrekan, yang dapat menyebabkan kegagalan tugas dan latensi performa.
Masalah umum lainnya adalah latensi dan error parsing yang disebabkan oleh kompleksitas kode DAG. Misalnya, kode DAG yang berisi variabel Airflow di tingkat teratas kode dapat menyebabkan penundaan parsing, kelebihan beban database, kegagalan penjadwalan, dan waktu tunggu DAG habis.
Dalam tutorial ini, Anda akan mendiagnosis DAG contoh dan mempelajari cara memecahkan masalah penjadwalan dan parsing, meningkatkan penjadwalan DAG, serta mengoptimalkan kode DAG dan konfigurasi lingkungan untuk meningkatkan performa.
Tujuan
Bagian ini mencantumkan tujuan untuk contoh dalam tutorial ini.
Contoh: Scheduler tidak berfungsi dan latensi disebabkan oleh konkurensi tugas yang tinggi
Upload DAG contoh yang berjalan beberapa kali secara bersamaan dan diagnosis masalah latensi dan malfungsi penjadwal dengan Cloud Monitoring.
Optimalkan kode DAG Anda dengan menggabungkan tugas dan mengevaluasi dampak performa.
Mendistribusikan tugas secara lebih merata dari waktu ke waktu dan mengevaluasi dampak performa.
Optimalkan konfigurasi Airflow dan konfigurasi lingkungan Anda serta evaluasi dampaknya.
Contoh: Error penguraian DAG dan latensi yang disebabkan oleh kode yang kompleks
Upload DAG sampel dengan variabel Airflow dan diagnosis masalah parsing dengan Cloud Monitoring.
Optimalkan kode DAG dengan menghindari variabel Airflow di tingkat teratas kode dan evaluasi dampaknya terhadap waktu penguraian.
Mengoptimalkan konfigurasi Airflow dan konfigurasi lingkungan serta mengevaluasi dampaknya terhadap waktu parsing.
Biaya
Tutorial ini menggunakan komponen Google Cloud yang dapat ditagih berikut: Google Cloud
Setelah menyelesaikan tutorial ini, Anda dapat menghindari penagihan berkelanjutan dengan menghapus resource yang Anda buat. Untuk mengetahui detail selengkapnya, lihat Pembersihan.
Sebelum memulai
Bagian ini menjelaskan tindakan yang diperlukan sebelum Anda memulai tutorial.
Membuat dan mengonfigurasi project
Untuk tutorial ini, Anda memerlukan Google Cloud project. Konfigurasi project dengan cara berikut:
Di Google Cloud konsol, pilih atau buat project:
Pastikan penagihan diaktifkan untuk project Anda. Pelajari cara memeriksa apakah penagihan telah diaktifkan pada suatu project.
Pastikan pengguna project Anda memiliki peran berikut untuk membuat resource yang diperlukan: Google Cloud
- Environment and Storage Object Administrator
(
roles/composer.environmentAndStorageObjectAdmin) - Compute Admin (
roles/compute.admin)
- Environment and Storage Object Administrator
(
Aktifkan API untuk project Anda
Enable the Cloud Composer API.
Buat lingkungan Cloud Composer Anda
Buat lingkungan Cloud Composer 2.
Sebagai bagian dari pembuatan lingkungan, Anda memberikan peran Ekstensi Agen Layanan Cloud Composer v2 API (roles/composer.ServiceAgentV2Ext) ke akun Agen Layanan Composer. Cloud Composer menggunakan akun ini untuk melakukan operasi
di project Google Cloud Anda.
Contoh: Penjadwal tidak berfungsi dan tugas gagal karena masalah penjadwalan tugas
Contoh ini menunjukkan cara men-debug malfungsi penjadwal dan latensi yang disebabkan oleh konkurensi tugas yang tinggi.
Mengupload DAG sampel ke lingkungan Anda
Upload contoh DAG berikut ke lingkungan
yang Anda buat pada langkah sebelumnya. Dalam tutorial ini, DAG ini diberi nama
dag_10_tasks_200_seconds_1.
DAG ini memiliki 200 tugas. Setiap tugas menunggu selama 1 detik dan mencetak "Selesai!". DAG dipicu secara otomatis setelah diupload. Cloud Composer menjalankan DAG ini 10 kali, dan semua eksekusi DAG terjadi secara paralel.
import time
from datetime import datetime, timedelta
from airflow import DAG
from airflow.decorators import task
tasks_amount = 200
seconds = 1
minutes = 5
with DAG(
dag_id=f"dag_10_tasks_{tasks_amount}_sec_{seconds}",
start_date=datetime(2023, 11, 22, 20, 0),
end_date=datetime(2023, 11, 22, 20, 49),
schedule_interval=timedelta(minutes=minutes),
catchup=True,
) as dag:
@task
def create_subtasks(seconds: int) -> None:
time.sleep(seconds)
for i in range(tasks_amount):
create_subtasks(seconds)
Mendiagnosis masalah malfungsi penjadwal dan kegagalan tugas
Setelah DAG selesai berjalan, buka UI Airflow, lalu klik DAG dag_10_tasks_200_seconds_1. Anda akan melihat bahwa total 10 operasi DAG berhasil, dan setiap operasi memiliki 200 tugas yang berhasil.
Tinjau log tugas Airflow:
Di konsol Google Cloud , buka halaman Environments.
Dalam daftar lingkungan, klik nama lingkungan Anda. Halaman Environment details akan terbuka.
Buka tab Logs, lalu buka All logs > Airflow logs > Workers > View in Logs Explorer.
Pada histogram log, Anda dapat melihat error dan peringatan yang ditunjukkan dengan warna merah dan oranye:
DAG contoh menghasilkan sekitar 130 peringatan dan 60 error. Klik kolom yang berisi batang kuning dan merah. Anda akan melihat beberapa peringatan dan error berikut dalam log:
State of this instance has been externally set to success. Terminating
instance.
Received SIGTERM. Terminating subprocesses.
worker: Warm shutdown (MainProcess).
Log ini mungkin menunjukkan bahwa penggunaan resource melampaui batas dan pekerja memulai ulang dirinya sendiri.
Jika tugas Airflow tetap berada dalam antrean terlalu lama, penjadwal akan menandainya sebagai gagal dan up_for_retry, lalu menjadwalkannya ulang untuk dieksekusi. Salah satu cara untuk mengamati gejala situasi ini adalah dengan melihat diagram dengan jumlah tugas dalam antrean dan jika lonjakan dalam diagram ini tidak turun dalam waktu sekitar 10 menit, kemungkinan akan terjadi kegagalan tugas (tanpa log).
Tinjau informasi pemantauan:
Buka tab Monitoring, lalu pilih Ringkasan.
Tinjau grafik tugas Airflow.
Gambar 2. Grafik tugas Airflow (klik untuk memperbesar) Dalam grafik tugas Airflow, terdapat lonjakan tugas dalam antrean yang berlangsung selama lebih dari 10 menit, yang mungkin berarti tidak ada cukup resource di lingkungan Anda untuk memproses semua tugas terjadwal.
Tinjau grafik Pekerja aktif:
Gambar 3. Grafik pekerja aktif (klik untuk memperbesar) Grafik Pekerja aktif menunjukkan bahwa DAG memicu penskalaan otomatis hingga batas maksimum yang diizinkan, yaitu tiga pekerja selama proses DAG.
Grafik penggunaan resource dapat menunjukkan kurangnya kapasitas di worker Airflow untuk menjalankan tugas yang diantrekan. Di tab Monitoring, pilih Pekerja dan tinjau grafik Total penggunaan CPU pekerja dan Total penggunaan memori pekerja.
Gambar 4. Grafik penggunaan CPU total pekerja (klik untuk memperbesar) 
Gambar 5. Grafik total penggunaan memori pekerja (klik untuk memperbesar) Grafik menunjukkan bahwa eksekusi terlalu banyak tugas secara bersamaan mengakibatkan batas CPU tercapai. Resource telah digunakan selama lebih dari 30 menit, yang bahkan lebih lama dari total durasi 200 tugas dalam 10 run DAG yang berjalan satu per satu.
Berikut adalah indikator bahwa antrean sudah penuh dan tidak ada cukup resource untuk memproses semua tugas terjadwal.
Menggabungkan tugas Anda
Kode saat ini membuat banyak DAG dan tugas tanpa resource yang memadai untuk memproses semua tugas secara paralel, sehingga menyebabkan antrean terisi. Menyimpan tugas dalam antrean terlalu lama dapat menyebabkan tugas dijadwalkan ulang atau gagal. Dalam situasi seperti ini, Anda harus memilih lebih sedikit tugas yang lebih terkonsolidasi.
DAG contoh berikut mengubah jumlah tugas dalam contoh awal dari 200 menjadi 20 dan meningkatkan waktu tunggu dari 1 menjadi 10 detik untuk meniru lebih banyak tugas gabungan yang melakukan jumlah pekerjaan yang sama.
Upload contoh DAG berikut ke lingkungan
yang Anda buat. Dalam tutorial ini, DAG ini diberi nama
dag_10_tasks_20_seconds_10.
import time
from datetime import datetime, timedelta
from airflow import DAG
from airflow.decorators import task
tasks_amount = 20
seconds = 10
minutes = 5
with DAG(
dag_id=f"dag_10_tasks_{tasks_amount}_sec_{seconds}",
start_date=datetime(2021, 12, 22, 20, 0),
end_date=datetime(2021, 12, 22, 20, 49),
schedule_interval=timedelta(minutes=minutes),
catchup=True,
) as dag:
@task
def create_subtasks(seconds: int) -> None:
time.sleep(seconds)
for i in range(tasks_amount):
create_subtasks(seconds)
Mengevaluasi dampak tugas yang lebih terkonsolidasi pada proses penjadwalan:
Tunggu hingga proses DAG selesai.
Di UI Airflow, di halaman DAG, klik DAG
dag_10_tasks_20_seconds_10. Anda akan melihat 10 eksekusi DAG, yang masing-masing memiliki 20 tugas yang berhasil.Di konsol Google Cloud , buka halaman Environments.
Dalam daftar lingkungan, klik nama lingkungan Anda. Halaman Environment details akan terbuka.
Buka tab Logs, lalu buka All logs > Airflow logs > Workers > View in Logs Explorer.
Contoh kedua dengan tugas yang lebih digabungkan menghasilkan sekitar 10 peringatan dan 7 error. Pada histogram, Anda dapat membandingkan jumlah error dan peringatan dalam contoh awal (nilai sebelumnya) dan contoh kedua (nilai selanjutnya).
Gambar 6. Histogram log worker Airflow setelah tugas digabungkan (klik untuk memperbesar) Saat membandingkan contoh pertama dengan contoh yang lebih digabungkan, Anda dapat melihat bahwa ada lebih sedikit error dan peringatan dalam contoh kedua. Namun, error yang sama terkait penonaktifan sementara masih muncul di log karena kelebihan beban resource.
Di tab Monitoring, pilih Pekerja dan tinjau grafik.
Saat membandingkan grafik tugas Airflow untuk contoh pertama (nilai yang lebih awal) dengan grafik untuk contoh kedua dengan tugas yang lebih digabungkan, Anda dapat melihat bahwa lonjakan tugas dalam antrean berlangsung dalam jangka waktu yang lebih singkat saat tugas lebih digabungkan. Namun, proses ini berlangsung hampir 10 menit, yang masih kurang optimal.
Gambar 7. Grafik tugas Airflow setelah tugas digabungkan (klik untuk memperbesar) Pada grafik Pekerja aktif, Anda dapat melihat contoh pertama (di sisi kiri grafik) menggunakan resource untuk jangka waktu yang jauh lebih lama daripada contoh kedua, meskipun kedua contoh meniru jumlah pekerjaan yang sama.
Gambar 8. Grafik pekerja aktif setelah tugas digabungkan (klik untuk memperbesar) Tinjau grafik konsumsi resource pekerja. Meskipun perbedaan antara resource yang digunakan dalam contoh dengan tugas yang lebih digabungkan dan contoh awal cukup signifikan, penggunaan CPU masih melonjak hingga 70% dari batas.
Gambar 9. Grafik penggunaan CPU pekerja total setelah tugas digabungkan (klik untuk memperbesar) 
Gambar 10. Grafik total penggunaan memori pekerja setelah tugas digabungkan (klik untuk memperbesar)
Mendistribusikan tugas secara lebih merata dari waktu ke waktu
Terlalu banyak tugas serentak akan menyebabkan antrean terisi, yang menyebabkan tugas macet dalam antrean atau dijadwalkan ulang. Pada langkah-langkah sebelumnya, Anda mengurangi jumlah tugas dengan menggabungkan tugas-tugas tersebut, tetapi log output dan pemantauan menunjukkan bahwa jumlah tugas serentak masih kurang optimal.
Anda dapat mengontrol jumlah tugas yang dijalankan secara serentak dengan menerapkan jadwal atau menetapkan batas jumlah tugas yang dapat dijalankan secara bersamaan.
Dalam tutorial ini, Anda akan mendistribusikan tugas secara lebih merata dari waktu ke waktu dengan menambahkan parameter tingkat DAG ke DAG dag_10_tasks_20_seconds_10:
Tambahkan argumen
max_active_runs=1ke pengelola konteks DAG. Argumen ini menetapkan batas hanya satu instance proses DAG yang berjalan pada waktu tertentu.Tambahkan argumen
max_active_tasks=5ke pengelola konteks DAG. Argumen ini mengontrol jumlah maksimum instance tugas yang dapat berjalan secara serentak di setiap DAG.
Upload contoh DAG berikut ke lingkungan
yang Anda buat. Dalam tutorial ini, DAG ini diberi nama
dag_10_tasks_20_seconds_10_scheduled.py.
import time
from datetime import datetime, timedelta
from airflow import DAG
from airflow.decorators import task
tasks_amount = 20
seconds = 10
minutes = 5
active_runs = 1
active_tasks = 5
with DAG(
dag_id=f"dag_10_tasks_{tasks_amount}_sec_{seconds}_runs_{active_runs}_tasks_{active_tasks}",
start_date=datetime(2021, 12, 22, 20, 0),
end_date=datetime(2021, 12, 22, 20, 49),
schedule_interval=timedelta(minutes=minutes),
max_active_runs=active_runs,
max_active_tasks=active_tasks,
catchup=True,
) as dag:
@task
def create_subtasks(seconds: int) -> None:
time.sleep(seconds)
for i in range(tasks_amount):
create_subtasks(seconds)
Mengevaluasi dampak pendistribusian tugas dari waktu ke waktu pada proses penjadwalan:
Tunggu hingga proses DAG selesai.
Di konsol Google Cloud , buka halaman Environments.
Dalam daftar lingkungan, klik nama lingkungan Anda. Halaman Environment details akan terbuka.
Buka tab Logs, lalu buka All logs > Airflow logs > Workers > View in Logs Explorer.
Pada histogram, Anda dapat melihat bahwa DAG ketiga dengan jumlah tugas dan proses aktif yang terbatas tidak menghasilkan peringatan atau error apa pun dan distribusi log terlihat lebih merata dibandingkan dengan nilai sebelumnya.
Gambar 11. Histogram log worker Airflow setelah tugas digabungkan dan didistribusikan dari waktu ke waktu (klik untuk memperbesar)
Tugas dalam contoh dag_10_tasks_20_seconds_10_scheduled yang memiliki
jumlah tugas aktif dan eksekusi yang terbatas tidak menyebabkan tekanan resource karena
tugas diantrekan secara merata.
Setelah melakukan langkah-langkah yang dijelaskan, Anda mengoptimalkan penggunaan resource dengan menggabungkan tugas-tugas kecil dan mendistribusikannya secara lebih merata dari waktu ke waktu.
Mengoptimalkan konfigurasi lingkungan
Anda dapat menyesuaikan konfigurasi lingkungan untuk memastikan selalu ada kapasitas di pekerja Airflow untuk menjalankan tugas yang diantrekan.
Jumlah pekerja dan konkurensi pekerja
Anda dapat menyesuaikan jumlah maksimum pekerja agar Cloud Composer otomatis menskalakan lingkungan Anda dalam batas yang ditetapkan.
Parameter [celery]worker_concurrency menentukan jumlah maksimum tugas yang dapat diambil oleh satu pekerja dari antrean tugas. Mengubah parameter ini akan menyesuaikan jumlah tugas yang dapat dijalankan oleh satu pekerja secara bersamaan.
Anda dapat mengubah opsi konfigurasi Airflow ini dengan
menggantinya. Secara default, konkurensi pekerja ditetapkan berdasarkan jumlah instance tugas serentak ringan yang dapat ditampung oleh pekerja. Artinya, nilainya bergantung pada batas resource pekerja.
Nilai konkurensi pekerja tidak bergantung pada jumlah pekerja di
lingkungan Anda.
Jumlah pekerja dan konkurensi pekerja bekerja bersama-sama, dan performa lingkungan Anda sangat bergantung pada kedua parameter tersebut. Anda dapat menggunakan pertimbangan berikut untuk memilih kombinasi yang tepat:
Beberapa tugas cepat yang berjalan secara paralel. Anda dapat meningkatkan konkurensi worker saat ada tugas yang menunggu dalam antrean, dan worker Anda menggunakan persentase CPU dan memori yang rendah pada saat yang sama. Namun, dalam keadaan tertentu, antrean mungkin tidak pernah terisi, sehingga penskalaan otomatis tidak pernah dipicu. Jika tugas kecil selesai dieksekusi pada saat pekerja baru siap, pekerja yang ada dapat mengambil tugas yang tersisa, dan tidak akan ada tugas untuk pekerja yang baru dibuat.
Dalam situasi ini, sebaiknya tingkatkan jumlah minimum pekerja dan tingkatkan konkurensi pekerja untuk menghindari penskalaan yang terlalu cepat.
Beberapa tugas panjang yang berjalan secara paralel. Konkurensi pekerja yang tinggi mencegah sistem menskalakan jumlah pekerja. Jika beberapa tugas membutuhkan banyak resource dan memerlukan waktu lama untuk diselesaikan, konkurensi pekerja yang tinggi dapat menyebabkan antrean tidak pernah terisi dan semua tugas hanya diambil oleh satu pekerja, yang mengakibatkan masalah performa. Dalam situasi ini, sebaiknya tingkatkan jumlah maksimum pekerja dan kurangi konkurensi pekerja.
Pentingnya paralelisme
Scheduler Airflow mengontrol penjadwalan eksekusi DAG dan tugas individual dari
DAG. Opsi konfigurasi Airflow [core]parallelism mengontrol jumlah
tugas yang dapat diantrekan oleh scheduler Airflow di antrean executor setelah semua
dependensi untuk tugas ini terpenuhi.
Paralelisme adalah mekanisme perlindungan Airflow yang menentukan jumlah tugas yang dapat dijalankan secara bersamaan per scheduler, terlepas dari jumlah pekerja. Nilai paralelisme, dikalikan dengan jumlah penjadwal di cluster Anda, adalah jumlah maksimum instance tugas yang dapat diantrekan oleh lingkungan Anda.
Biasanya, [core]parallelism ditetapkan sebagai hasil perkalian jumlah maksimum pekerja dan [celery]worker_concurrency. Hal ini juga dipengaruhi oleh
pool.
Anda dapat mengubah opsi konfigurasi Airflow ini dengan
menggantinya. Untuk mengetahui informasi selengkapnya tentang penyesuaian konfigurasi Airflow terkait penskalaan, lihat Konfigurasi penskalaan Airflow.
Menemukan konfigurasi lingkungan yang optimal
Cara yang direkomendasikan untuk memperbaiki masalah penjadwalan adalah menggabungkan tugas kecil menjadi tugas yang lebih besar dan mendistribusikan tugas secara lebih merata dari waktu ke waktu. Selain mengoptimalkan kode DAG, Anda juga dapat mengoptimalkan konfigurasi lingkungan agar memiliki kapasitas yang memadai untuk menjalankan beberapa tugas secara bersamaan.
Misalnya, asumsikan Anda menggabungkan tugas dalam DAG sebanyak mungkin, tetapi membatasi tugas aktif untuk menyebarkannya secara lebih merata dari waktu ke waktu bukanlah solusi yang disukai untuk kasus penggunaan spesifik Anda.
Anda dapat menyesuaikan paralelisme, jumlah pekerja, dan parameter
serentak pekerja untuk menjalankan DAG dag_10_tasks_20_seconds_10 tanpa membatasi tugas
aktif. Dalam contoh ini, DAG berjalan 10 kali dan setiap kali berjalan berisi 20 tugas kecil.
Jika Anda ingin menjalankan semuanya secara bersamaan:
Anda akan memerlukan ukuran lingkungan yang lebih besar, karena ukuran ini mengontrol parameter performa infrastruktur Cloud Composer terkelola di lingkungan Anda.
Worker Airflow harus dapat menjalankan 20 tugas secara bersamaan, yang berarti Anda perlu menyetel konkurensi worker ke 20.
Pekerja memerlukan CPU dan memori yang memadai untuk menangani semua tugas. Konkurensi worker dipengaruhi oleh CPU dan memori worker, sehingga Anda memerlukan setidaknya
worker_concurrency / 12CPU danleast worker_concurrency / 8memori.Anda harus meningkatkan paralelisme agar sesuai dengan konkurensi pekerja yang lebih tinggi. Agar pekerja dapat mengambil 20 tugas dari antrean, penjadwal harus menjadwalkan 20 tugas tersebut terlebih dahulu.
Sesuaikan konfigurasi lingkungan Anda dengan cara berikut:
Di konsol Google Cloud , buka halaman Environments.
Dalam daftar lingkungan, klik nama lingkungan Anda. Halaman Environment details akan terbuka.
Buka tab Konfigurasi lingkungan.
Temukan konfigurasi Resources > Workloads, lalu klik Edit.
Di bagian Worker, di kolom Memory, tentukan batas memori baru untuk pekerja Airflow. Dalam tutorial ini, gunakan 4 GB.
Di kolom CPU, tentukan batas CPU baru untuk pekerja Airflow. Dalam tutorial ini, gunakan 2 vCPU.
Simpan perubahan dan tunggu beberapa menit agar pekerja Airflow Anda dimulai ulang.
Selanjutnya, ganti opsi konfigurasi Airflow paralelisme dan konkurensi pekerja:
Buka tab Airflow Configuration Overrides.
Klik Edit, lalu klik Tambahkan Penggantian Konfigurasi Airflow.
Ganti konfigurasi paralelisme:
Bagian Kunci Nilai coreparallelism20Klik Add Airflow Configuration Override dan ganti konfigurasi konkurensi pekerja:
Bagian Kunci Nilai celeryworker_concurrency20Klik Simpan dan tunggu hingga lingkungan memperbarui konfigurasinya.
Aktifkan contoh DAG yang sama lagi dengan konfigurasi yang disesuaikan:
Di UI Airflow, buka halaman DAGs.
Temukan DAG
dag_10_tasks_20_seconds_10, lalu hapus.Setelah DAG dihapus, Airflow akan memeriksa folder DAG di bucket lingkungan Anda dan menjalankan DAG lagi secara otomatis.
Setelah DAG berjalan selesai, tinjau kembali histogram Log. Pada diagram,
Anda dapat melihat bahwa contoh dag_10_tasks_20_seconds_10 dengan lebih banyak
tugas yang digabungkan tidak menghasilkan error dan peringatan saat dijalankan dengan
konfigurasi lingkungan yang disesuaikan. Bandingkan hasilnya dengan data sebelumnya
pada diagram, tempat contoh yang sama menghasilkan error dan peringatan saat
dijalankan dengan konfigurasi lingkungan default.
Konfigurasi lingkungan dan konfigurasi Airflow memainkan peran penting dalam penjadwalan tugas, tetapi konfigurasi tidak dapat ditingkatkan melampaui batas tertentu.
Sebaiknya optimalkan kode DAG, gabungkan tugas, dan gunakan penjadwalan untuk performa dan efisiensi yang optimal.
Contoh: Error penguraian DAG dan latensi karena kode DAG yang kompleks
Dalam contoh ini, Anda akan menyelidiki latensi parsing DAG sampel yang meniru kelebihan variabel Airflow.
Buat variabel Airflow baru
Sebelum mengupload kode contoh, buat variabel Airflow baru.
Di konsol Google Cloud , buka halaman Environments.
Di kolom Airflow webserver, ikuti link Airflow untuk lingkungan Anda.
Buka Admin > Variabel > Tambahkan rekaman baru.
Tetapkan nilai berikut:
- kunci:
example_var - val:
test_airflow_variable
- kunci:
Mengupload DAG sampel ke lingkungan Anda
Upload contoh DAG berikut ke lingkungan
yang Anda buat pada langkah sebelumnya. Dalam tutorial ini, DAG ini diberi nama
dag_for_loop_airflow_variable.
DAG ini berisi loop for yang berjalan 1.000 kali dan meniru kelebihan variabel Airflow. Setiap iterasi membaca variabel example_var dan
membuat tugas. Setiap tugas berisi satu perintah yang mencetak nilai
variabel.
from datetime import datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.models import Variable
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2023, 2, 17),
'retries': 0
}
dag = DAG(
'dag_for_loop_airflow_variable',
default_args=default_args,
catchup=False,
schedule_interval="@daily"
)
for i in range(1000):
a = Variable.get('example_var', 'N/A')
task = BashOperator(
task_id=f'task_{i}',
bash_command='echo variable foo=${foo_env}',
dag=dag,
env={'foo_env': a}
)
Mendiagnosis masalah penguraian
Waktu penguraian DAG adalah jumlah waktu yang dibutuhkan scheduler Airflow untuk membaca dan menguraikan file DAG. Sebelum scheduler Airflow dapat menjadwalkan tugas apa pun dari DAG, scheduler harus mengurai file DAG untuk menemukan struktur DAG dan tugas yang ditentukan.
Jika DAG memerlukan waktu lama untuk diuraikan, hal ini akan menggunakan kapasitas penjadwal dan dapat mengurangi performa eksekusi DAG.
Untuk memantau waktu penguraian DAG:
Jalankan perintah Airflow CLI
dags reportdi gcloud CLI untuk melihat waktu parsing semua DAG Anda:gcloud composer environments run ENVIRONMENT_NAME \ --location LOCATION \ dags reportGanti kode berikut:
ENVIRONMENT_NAME: nama lingkungan Anda.LOCATION: region tempat lingkungan berada.
Di output perintah, cari nilai durasi untuk DAG
dag_for_loop_airflow_variables. Nilai yang besar dapat menunjukkan bahwa DAG ini tidak diimplementasikan secara optimal. Jika memiliki beberapa DAG, dari tabel output, Anda dapat mengidentifikasi DAG mana yang memiliki waktu parsing yang lama.Contoh:
file | duration | dag_num | task_num | dags ====================+================+=========+==========+===================== /dag_for_loop_airfl | 0:00:14.773594 | 1 | 1000 | dag_for_loop_airflow ow_variable.py | | | | _variable /airflow_monitoring | 0:00:00.003035 | 1 | 1 | airflow_monitoring .pyPeriksa waktu parsing DAG di konsol Google Cloud :
- Di konsol Google Cloud , buka halaman Environments.
Dalam daftar lingkungan, klik nama lingkungan Anda. Halaman Environment details akan terbuka.
Buka tab Logs, lalu buka All logs > DAG processor manager.
Tinjau log
dag-processor-managerdan identifikasi kemungkinan masalah.
Gambar 13. Log pengelola pemroses DAG menampilkan waktu penguraian DAG (klik untuk memperbesar)
Jika total waktu penguraian DAG melebihi sekitar 10 detik, penjadwal Anda mungkin kelebihan beban dengan penguraian DAG dan tidak dapat menjalankan DAG secara efektif.
Mengoptimalkan kode DAG
Sebaiknya hindari kode Python "level teratas" yang tidak perlu dalam DAG Anda. DAG dengan banyak impor, variabel, dan fungsi di luar DAG akan menyebabkan waktu parsing yang lebih lama untuk penjadwal Airflow. Hal ini mengurangi performa dan skalabilitas Cloud Composer dan Airflow. Membaca variabel Airflow yang berlebihan menyebabkan waktu parsing yang lama dan beban database yang tinggi. Jika kode ini ada dalam file DAG, fungsi ini akan dieksekusi pada setiap detak jantung penjadwal, yang mungkin lambat.
Kolom template Airflow memungkinkan Anda menyertakan nilai dari variabel Airflow dan template Jinja ke dalam DAG. Tindakan ini mencegah eksekusi fungsi yang tidak perlu selama detak jantung penjadwal.
Untuk menerapkan contoh DAG dengan cara yang lebih baik, hindari penggunaan variabel Airflow di kode Python tingkat teratas DAG. Sebagai gantinya, teruskan variabel Airflow ke operator yang ada melalui template Jinja, yang akan menunda pembacaan nilai hingga eksekusi tugas.
Upload DAG contoh versi baru ke lingkungan Anda. Dalam tutorial ini, DAG ini diberi nama
dag_for_loop_airflow_variable_optimized.
from datetime import datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2023, 2, 17),
'retries': 0
}
dag = DAG(
'dag_for_loop_airflow_variable_optimized',
default_args=default_args,
catchup=False,
schedule_interval='@daily'
)
for i in range(1000):
task = BashOperator(
task_id=f'bash_use_variable_good_{i}',
bash_command='echo variable foo=${foo_env}',
dag=dag,
env={'foo_env': '{{ var.value.get("example_var") }}'},
)
Periksa waktu penguraian DAG baru:
Tunggu hingga proses DAG selesai.
Jalankan perintah
dags reportlagi untuk melihat waktu penguraian semua DAG Anda:file | duration | dag_num | task_num | dags ====================+================+=========+==========+===================== /dag_for_loop_airfl | 0:00:37.000369 | 1 | 1000 | dag_for_loop_airflow ow_variable.py | | | | _variable /dag_for_loop_airfl | 0:00:01.109457 | 1 | 1000 | dag_for_loop_airflow ow_variable_optimiz | | | | _variable_optimized ed.py | | | | /airflow_monitoring | 0:00:00.040510 | 1 | 1 | airflow_monitoring .py | | | |Tinjau kembali
dag-processor-managerlog dan analisis durasi penguraian.
Gambar 14. Log pengelola pemroses DAG menampilkan waktu penguraian DAG setelah kode DAG dioptimalkan (klik untuk memperbesar)
Dengan mengganti variabel lingkungan dengan template Airflow, Anda menyederhanakan kode DAG dan mengurangi latensi parsing sekitar sepuluh kali lipat.
Mengoptimalkan konfigurasi lingkungan Airflow
Scheduler Airflow terus-menerus mencoba memicu tugas baru dan mengurai semua DAG di bucket lingkungan Anda. Jika DAG Anda memiliki waktu parsing yang lama dan scheduler menggunakan banyak resource, Anda dapat mengoptimalkan konfigurasi scheduler Airflow agar scheduler menggunakan resource secara lebih efisien.
Dalam tutorial ini, file DAG memerlukan banyak waktu untuk diuraikan, dan siklus penguraian mulai tumpang-tindih, yang kemudian menghabiskan kapasitas penjadwal. Dalam contoh kami,
DAG contoh pertama memerlukan waktu lebih dari 5 detik untuk diuraikan, jadi Anda akan mengonfigurasi
penjadwal agar berjalan lebih jarang untuk menggunakan resource secara lebih efisien. Anda akan mengganti opsi konfigurasi Airflow scheduler_heartbeat_sec. Konfigurasi ini menentukan seberapa sering penjadwal harus berjalan (dalam detik). Secara default, nilai ditetapkan ke 5 detik.
Anda dapat mengubah opsi konfigurasi Airflow ini dengan
menggantinya.
Ganti opsi konfigurasi Airflow scheduler_heartbeat_sec:
Di konsol Google Cloud , buka halaman Environments.
Dalam daftar lingkungan, klik nama lingkungan Anda. Halaman Environment details akan terbuka.
Buka tab Airflow Configuration Overrides.
Klik Edit, lalu klik Tambahkan Penggantian Konfigurasi Airflow.
Ganti opsi konfigurasi Airflow:
Bagian Kunci Nilai schedulerscheduler_heartbeat_sec10Klik Simpan dan tunggu hingga lingkungan memperbarui konfigurasinya.
Periksa metrik penjadwal:
Buka tab Monitoring, lalu pilih Schedulers.
Di grafik Detak jantung penjadwal, klik tombol Opsi lainnya (tiga titik), lalu klik Lihat di Metrics Explorer.
Pada grafik, Anda akan melihat penjadwal berjalan dua kali lebih jarang setelah Anda mengubah konfigurasi default dari 5 detik menjadi 10 detik. Dengan mengurangi frekuensi detak jantung, Anda memastikan bahwa penjadwal tidak mulai berjalan saat siklus parsing sebelumnya sedang berlangsung dan kapasitas resource penjadwal tidak habis.
Menetapkan lebih banyak resource ke penjadwal
Di Cloud Composer 2, Anda dapat mengalokasikan lebih banyak resource CPU dan memori ke penjadwal. Dengan cara ini, Anda dapat meningkatkan performa penjadwal dan mempercepat waktu parsing untuk DAG.
Mengalokasikan CPU dan memori tambahan ke penjadwal:
Di konsol Google Cloud , buka halaman Environments.
Dalam daftar lingkungan, klik nama lingkungan Anda. Halaman Environment details akan terbuka.
Buka tab Konfigurasi lingkungan.
Temukan konfigurasi Resources > Workloads, lalu klik Edit.
Di bagian Penjadwal, di kolom Memori, tentukan batas memori baru. Dalam tutorial ini, gunakan 4 GB.
Di kolom CPU, tentukan batas CPU baru. Dalam tutorial ini, gunakan 2 vCPU.
Simpan perubahan dan tunggu beberapa menit agar penjadwal Airflow Anda dimulai ulang.
Buka tab Logs, lalu buka All logs > DAG processor manager.
Tinjau log
dag-processor-managerdan bandingkan durasi penguraian untuk DAG contoh:
Gambar 16. Log pengelola pemroses DAG menampilkan waktu parsing DAG setelah lebih banyak resource ditetapkan ke penjadwal (klik untuk memperbesar)
Dengan menetapkan lebih banyak resource ke penjadwal, Anda meningkatkan kapasitas penjadwal dan mengurangi latensi parsing secara signifikan dibandingkan dengan konfigurasi lingkungan default. Dengan lebih banyak resource, penjadwal dapat mem-parsing DAG lebih cepat, tetapi biaya yang terkait dengan resource Cloud Composer juga akan meningkat. Selain itu, resource tidak dapat ditingkatkan melebihi batas tertentu.
Sebaiknya alokasikan resource hanya setelah kemungkinan pengoptimalan kode DAG dan konfigurasi Airflow diterapkan.
Pembersihan
Agar tidak dikenai biaya pada akun Google Cloud Anda untuk resource yang digunakan dalam tutorial ini, hapus project yang berisi resource tersebut, atau simpan project dan hapus setiap resource.
Menghapus project
- In the Google Cloud console, go to the Manage resources page.
- In the project list, select the project that you want to delete, and then click Delete.
- In the dialog, type the project ID, and then click Shut down to delete the project.
Menghapus resource satu per satu
Jika Anda berencana mempelajari beberapa tutorial dan panduan memulai, menggunakan kembali project dapat membantu Anda agar tidak melampaui batas kuota project.
Hapus lingkungan Cloud Composer. Anda juga menghapus bucket lingkungan selama prosedur ini.