KAL
Menu

Dynamic task mapping

Membentuk task saat runtime dari list dinamis dengan dynamic task mapping.

Daftar isi

Banyak pipeline memproses daftar item yang jumlahnya baru diketahui saat pipeline berjalan, misalnya daftar berkas harian yang belum lengkap ketika DAG ditulis. Dynamic task mapping menjawab kebutuhan itu dengan membentuk instans task saat runtime, yaitu satu instans untuk setiap elemen list yang dipetakan. Topik ini membahas penggunaan .expand() atas list statis maupun atas nilai balik task lain, penomoran instans terpetakan melalui map_index, serta pengujian DAG terpetakan dengan CLI. Contohnya adalah DAG yang memproses daftar berkas arsip per partisi.

Mengembangkan task dengan expand#

Pada gaya TaskFlow, sebuah task biasanya dipanggil satu kali sehingga membentuk satu task pada DAG. Pemanggilan biasa digantikan dengan metode .expand() untuk memetakan task atas sebuah list, dan setiap elemen list kemudian menjadi satu instans eksekusi tersendiri pada task yang sama. Jumlah task pada definisi DAG tetap satu, sedangkan jumlah instansnya baru ditentukan ketika dag run berlangsung. Bentuk paling sederhana memetakan task atas list yang ditulis langsung pada berkas DAG. Misalnya, proses.expand(berkas=["data-2026-10-01.csv", "data-2026-10-02.csv"]) membentuk dua instans proses.

Metode .expand() hanya menerima argumen kata kunci. Setiap argumen kata kunci berupa nama parameter fungsi task beserta list nilainya. Pola ini berlaku untuk tipe list apa pun yang dapat disimpan sebagai XCom, dan metode ini juga menerima beberapa argumen sekaligus sehingga menghasilkan kombinasi silang antar list.

List dari nilai balik task lain#

Dynamic task mapping juga dapat mengambil list dari nilai balik task lain, misalnya hasil pembacaan direktori atau panggilan API. Task yang menghasilkan list berjalan lebih dahulu, nilai baliknya disimpan sebagai XCom, kemudian task terpetakan dibentuk satu instans per elemen list itu. Jumlah instans kini mengikuti data, bukan kode.

Berkas berikut menerapkan pola tersebut. Task daftar_berkas menghasilkan dua nama berkas arsip, task proses dipetakan atas daftar itu, dan task rangkum menerima kumpulan hasil proses sebagai satu list utuh.

python partisi_arsip.py 
"""DAG yang memproses daftar berkas dengan dynamic task mapping."""
import pendulum
from airflow.sdk import dag, task


@dag(
    dag_id="partisi_arsip",
    schedule="@once",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
)
def partisi_arsip():
    @task
    def daftar_berkas() -> list[str]:
        return ["data-2026-10-01.csv", "data-2026-10-02.csv"]

    @task
    def proses(berkas: str) -> str:
        print(f"memproses {berkas}")
        return f"selesai: {berkas}"

    @task
    def rangkum(hasil: list[str]) -> None:
        print(f"{len(hasil)} berkas selesai diproses")

    rangkum(proses.expand(berkas=daftar_berkas()))


dag = partisi_arsip()

Baris rangkum(proses.expand(berkas=daftar_berkas())) sekaligus memperlihatkan dua mekanisme. Argumen berkas diisi XComArg dari daftar_berkas, sehingga pengembangan proses menunggu nilai balik task tersebut. Sementara itu, nilai balik proses yang terpetakan dikumpulkan menjadi satu list yang diterima rangkum sebagai parameter hasil. Pola pengumpulan ini membuat task fan-in, yaitu task yang menerima hasil dari banyak task hulu, tetap tunggal meskipun jumlah instansnya berubah-ubah.

Pada saat berkas DAG dibaca, task terpetakan belum bernomor dan hanya muncul sebagai satu task. Inspeksi berikut menunjukkan struktur DAG sebelum ada dag run.

bash 
# memeriksa struktur DAG terpetakan pada waktu perancangan
python3 -c "import partisi_arsip; d = partisi_arsip.dag; print(d.dag_id); print(len(d.tasks)); print(sorted(t.task_id for t in d.tasks))"
text Keluaran 
partisi_arsip
3
['daftar_berkas', 'proses', 'rangkum']

Task proses baru akan terpecah menjadi instans saat daftar_berkas selesai berjalan. Aturan ini berlaku ketat untuk konfigurasi yang bergantung pada data, termasuk trigger rule.

map_index dan instans terpetakan#

Setiap instans task terpetakan menerima indeks bilangan bulat yang disebut map_index, dimulai dari 0 mengikuti urutan elemen list sumbernya. Indeks ini menjadi bagian identitas setiap task instance, sehingga dua instans proses dapat dibedakan pada log, pada Grid view, dan pada basis data metadata meskipun task_id-nya sama. Log eksekusi memuat baris starting task_id=proses map_index=0 dan seterusnya untuk setiap instans yang berjalan.

Penomoran bawaan berupa angka memadai untuk daftar yang urutannya tidak bermakna. Apabila setiap instans perlu dikenali melalui namanya, misalnya nama berkas atau tanggal partisi, argumen map_index_template pada decorator @task memberi nama instans berdasarkan data input atau nilai yang ditetapkan selama eksekusi. Nama tersebut kemudian menggantikan angka pada UI. Saat membaca XCom, map_index juga dapat dirujuk agar hanya nilai satu instans yang diambil, misalnya saat menghitung atau melakukan debug pada instans tertentu.

Menjalankan DAG terpetakan#

Pengujian menggunakan AIRFLOW_HOME baru yang terpisah agar tidak bercampur dengan lingkungan latihan lain, lalu berkas DAG disalin ke folder dags dan basis data metadata disiapkan.

bash 
# menyiapkan AIRFLOW_HOME terpisah untuk latihan topik ini
export AIRFLOW_HOME=/tmp/ra-mapping
mkdir -p $AIRFLOW_HOME/dags
# menyalin berkas DAG ke folder dags
cp partisi_arsip.py $AIRFLOW_HOME/dags/
bash 
# menyiapkan basis data metadata Airflow
airflow db migrate > /dev/null 2>&1
echo "kode keluar db migrate: $?"
text Keluaran 
kode keluar db migrate: 0
bash 
# menjalankan seluruh task DAG partisi_arsip secara lokal
airflow dags test partisi_arsip > /tmp/ra-mapping/dags-test.log 2>&1
echo "kode keluar dags test: $?"
text Keluaran 
kode keluar dags test: 0

Log pada /tmp/ra-mapping/dags-test.log memperlihatkan pembentukan instans yang belum terlihat pada inspeksi sebelumnya. Task daftar_berkas berjalan lebih dahulu, kemudian proses dijalankan dua kali dengan map_index=0 dan map_index=1, satu instans untuk setiap nama berkas pada daftar sumber. Log instans pertama memuat baris memproses data-2026-10-01.csv, sedangkan log instans kedua memuat baris memproses data-2026-10-02.csv. Terakhir, rangkum berjalan satu kali dan mencetak 2 berkas selesai diproses, yang menandakan hasil kedua instans terpetakan terkumpul sebagai satu list. Dag run berakhir dengan status successful.

Apabila daftar sumber diperpanjang menjadi lima berkas, tidak ada satu baris kode pun yang berubah. Jumlah instans proses menjadi lima pada dag run berikutnya, dan rangkum otomatis menerima list dengan lima anggota.

Kesalahan umum#

Memanggil task tanpa .expand(). Pemanggilan proses(daftar_berkas()) yang dimaksudkan untuk memetakan justru membentuk satu task biasa yang menerima seluruh list sebagai satu argumen, tanpa pengembangan instans. Gunakan proses.expand(berkas=...) agar setiap elemen mendapat instans sendiri.

Memberi argumen posisional kepada .expand(). Metode ini hanya menerima argumen kata kunci, sehingga proses.expand(daftar) ditolak dengan error. Sebut nama parameternya secara eksplisit sesuai tanda tangan fungsi task.

Menghitung instans terpetakan pada saat perancangan. Daftar task pada saat berkas DAG dibaca masih memuat proses tanpa nomor, sehingga logika yang bergantung pada jumlah instans harus ditulis di dalam task, bukan di tingkat berkas DAG.

Ringkasan#

  • .expand() memetakan satu task atas sebuah list dan membentuk satu instans eksekusi per elemen saat runtime.
  • List sumber boleh ditulis langsung pada DAG atau dihasilkan task lain melalui XComArg, sehingga jumlah instans mengikuti data.
  • Nilai balik task terpetakan dikumpulkan menjadi satu list untuk task hilir yang menerimanya sebagai argumen.
  • Setiap instans diberi map_index mulai dari 0, dan dapat diberi nama melalui map_index_template.

Referensi#