TaskFlow API

TaskFlow API adalah gaya penulisan DAG modern pada Apache Airflow yang menggunakan decorator Python alih-alih membuat objek operator satu per satu. Topik ini membahas decorator @task dan @dag, cara nilai balik sebuah task menjadi XCom, serta cara dependensi antar task disimpulkan otomatis dari pemanggilan fungsinya. Dua versi DAG yang setara, yaitu versi klasik dengan PythonOperator dan versi TaskFlow, dibandingkan agar perbedaan keduanya terlihat jelas. Setelah topik ini, DAG sederhana dapat ditulis dengan gaya TaskFlow tanpa menyambungkan XCom secara manual.

Task pertama dengan decorator @task#

Decorator @task mengubah sebuah fungsi Python biasa menjadi task Airflow, yaitu satuan kerja yang dijadwalkan dan dijalankan oleh scheduler. Fungsi yang didekorasi tetap ditulis seperti fungsi biasa, lengkap dengan parameter dan anotasi tipe untuk nilai baliknya, sedangkan Airflow mengurus pengiriman argumen dan penyimpanan hasil melalui XCom. Fitur ini tersedia sejak Airflow 2.0, sehingga cara penulisan task pada topik ini tetap berlaku pada versi 2.x maupun 3.x.

Berkas berikut mendefinisikan DAG statistik nilai ujian dengan tiga task, yaitu ambil_nilai yang mengembalikan daftar nilai, hitung_rata_rata yang menghitung rata-rata daftar nilai tersebut, dan tampilkan yang mencetak hasilnya.

"""DAG statistik nilai ujian dengan TaskFlow API."""
import pendulum
from airflow.sdk import dag, task


@dag(
    dag_id="statistik_taskflow",
    schedule="@once",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
)
def statistik_taskflow():
    @task
    def ambil_nilai() -> list[int]:
        return [72, 85, 90, 68, 79]

    @task
    def hitung_rata_rata(nilai: list[int]) -> float:
        return sum(nilai) / len(nilai)

    @task
    def tampilkan(rata_rata: float) -> None:
        print(f"Rata-rata kelas: {rata_rata:.2f}")

    tampilkan(hitung_rata_rata(ambil_nilai()))


dag = statistik_taskflow()

Baris tampilkan(hitung_rata_rata(ambil_nilai())) adalah inti dari gaya TaskFlow. Pada saat berkas DAG dibaca, pemanggilan ambil_nilai() tidak menjalankan fungsinya, tetapi menghasilkan XComArg, yaitu objek yang mewakili nilai XCom, dan nilai itu baru terisi saat task benar-benar berjalan. Objek itu menjadi argumen bagi hitung_rata_rata, sehingga Airflow sekaligus mencatat bahwa hitung_rata_rata merupakan task hilir (downstream) dari ambil_nilai. Mekanisme yang sama berlaku untuk tampilkan, sehingga rantai dependensi dan pengaliran data terbentuk hanya dari cara fungsi task itu dipanggil.

Untuk memastikan berkas DAG dapat diparse tanpa error, jalankan berkas DAG itu secara langsung. Berkas DAG yang benar tidak mencetak apa pun, sehingga perintah yang berjalan tanpa keluaran menandakan berkas siap dipindahkan ke folder dags. Uji sederhana ini menangkap error sintaks dan error impor lebih awal.

# berkas DAG tidak mencetak apa pun bila dapat diurai dengan benar
python3 statistik_taskflow.py

Struktur DAG hasil perancangan dapat diperiksa dengan mengimpor modulnya dan membaca daftar task.

# memeriksa struktur DAG hasil perancangan
python3 -c "import statistik_taskflow; d = statistik_taskflow.dag; print(d.dag_id); print(sorted(t.task_id for t in d.tasks))"

statistik_taskflow
['ambil_nilai', 'hitung_rata_rata', 'tampilkan']

Versi setara dengan PythonOperator#

Gaya klasik menyusun DAG dari instans operator, yaitu kelas yang merangkai satu unit kerja beserta cara eksekusinya. Data antar task dipindahkan secara eksplisit. PythonOperator secara otomatis menyimpan nilai balik callable sebagai XCom, sedangkan task berikutnya harus menarik nilai itu kembali melalui xcom_pull dengan menyebut task_ids task sumber. Dependensi antar operator dideklarasikan dengan operator bitshift >>, terpisah dari logika pemindahan data.

Berkas berikut adalah versi klasik dari DAG yang sama, dengan tiga task yang identik.

"""DAG statistik nilai ujian dengan gaya klasik PythonOperator."""
import pendulum
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG


def ambil_nilai():
    return [72, 85, 90, 68, 79]


def hitung_rata_rata(**context):
    nilai = context["ti"].xcom_pull(task_ids="ambil_nilai")
    return sum(nilai) / len(nilai)


def tampilkan(**context):
    rata_rata = context["ti"].xcom_pull(task_ids="hitung_rata_rata")
    print(f"Rata-rata kelas: {rata_rata:.2f}")


with DAG(
    dag_id="statistik_klasik",
    schedule="@once",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
) as dag:
    t_ambil = PythonOperator(task_id="ambil_nilai", python_callable=ambil_nilai)
    t_hitung = PythonOperator(
        task_id="hitung_rata_rata", python_callable=hitung_rata_rata
    )
    t_tampil = PythonOperator(task_id="tampilkan", python_callable=tampilkan)
    t_ambil >> t_hitung >> t_tampil

Perhatikan bahwa fungsi callable pada versi klasik menerima **context dan menarik XCom secara manual, sedangkan pada versi TaskFlow nilai itu datang sebagai parameter biasa dengan tipe yang jelas. Inspeksi berikut menunjukkan bahwa kedua DAG menghasilkan daftar task yang sama persis.

# memeriksa struktur DAG versi klasik
python3 -c "import statistik_klasik; d = statistik_klasik.dag; print(d.dag_id); print(sorted(t.task_id for t in d.tasks))"

statistik_klasik
['ambil_nilai', 'hitung_rata_rata', 'tampilkan']

Tabel berikut merangkum perbedaan kedua gaya pada DAG yang setara ini.

Decorator @dag dan parameter task#

Decorator @dag mengubah sebuah fungsi menjadi generator DAG, yaitu fungsi yang menghasilkan DAG setiap kali dipanggil. Fungsi yang didekorasi tidak langsung menjadi DAG, sehingga berkas di atas ditutup dengan pemanggilan dag = statistik_taskflow(). Argumen schedule, start_date, dan catchup pada decorator menggantikan pembuatan objek DAG di dalam blok with. Pendekatan ini juga berlaku untuk parameter, karena parameter fungsi generator otomatis menjadi parameter DAG yang dapat diisi saat DAG dipicu.

Nilai balik sebuah task tidak selalu berupa satu angka. Apabila fungsi mengembalikan dict, decorator @task dengan argumen multiple_outputs=True menyimpan setiap kuncinya sebagai XCom terpisah, sehingga task hilir dapat merujuk satu kunci saja.

# tidak-dijalankan
@task(multiple_outputs=True)
def ringkas(nilai: list[int]) -> dict[str, float]:
    return {"rata_rata": sum(nilai) / len(nilai), "maks": max(nilai)}

# kunci rata_rata dan maks tersedia sebagai dua XCom terpisah

Menjalankan DAG dan membaca aliran nilai#

Perintah airflow dags test menjalankan seluruh task dalam sebuah DAG secara lokal tanpa scheduler. Untuk latihan ini, AIRFLOW_HOME diarahkan ke direktori sementara yang terpisah, kemudian kedua berkas DAG disalin ke folder dags agar terbaca oleh CLI.

# menyiapkan AIRFLOW_HOME terpisah untuk latihan topik ini
export AIRFLOW_HOME=/tmp/ra-taskflow
mkdir -p $AIRFLOW_HOME/dags
# menyalin kedua berkas DAG ke folder dags
cp statistik_taskflow.py statistik_klasik.py $AIRFLOW_HOME/dags/

Sebelum DAG dapat diuji, basis data metadata Airflow perlu disiapkan. Keluaran perintah airflow db migrate panjang dan berisi stempel waktu, sehingga keluaran itu dialihkan dan cukup kode keluarnya yang diperiksa.

# menyiapkan basis data metadata Airflow
airflow db migrate > /dev/null 2>&1
echo "kode keluar db migrate: $?"

kode keluar db migrate: 0

Selanjutnya DAG TaskFlow dijalankan. Keluaran yang panjang dialihkan ke berkas log karena berisi stempel waktu, sedangkan status akhir diperiksa melalui kode keluar.

# menjalankan seluruh task DAG statistik_taskflow secara lokal
airflow dags test statistik_taskflow > /tmp/ra-taskflow/dags-test.log 2>&1
echo "kode keluar dags test: $?"

kode keluar dags test: 0

Log pada /tmp/ra-taskflow/dags-test.log memperlihatkan aliran nilai yang disimpulkan TaskFlow. Task berjalan berurutan sesuai dependensi, yaitu ambil_nilai, kemudian hitung_rata_rata, lalu tampilkan, dan dag run diakhiri dengan status successful. Log task tampilkan memuat baris Rata-rata kelas: 78.80, yang dihitung dari lima nilai pada daftar sumber. Seluruh pemindahan nilai terjadi tanpa satu pun pemanggilan xcom_pull di dalam kode. Menjalankan versi klasik dengan airflow dags test statistik_klasik menghasilkan urutan task dan baris keluaran yang sama, karena kedua DAG merupakan logika yang setara.

Kesalahan umum#

Menganggap nilai balik pemanggilan task sebagai nilai sungguhan. Pada saat berkas DAG dibaca, ambil_nilai() menghasilkan XComArg, bukan daftar bilangan, sehingga ekspresi seperti len(ambil_nilai()) gagal dengan error. Operasi atas nilai tersebut hanya dapat ditulis di dalam fungsi task lain yang menerima objek itu sebagai argumen.

Menghubungkan task tanpa pemanggilan. Fungsi berdecorator @task baru terdaftar pada DAG ketika fungsinya dipanggil di dalam generator. Membuat tiga fungsi berdecorator @task tetapi tidak pernah memanggilnya menghasilkan DAG tanpa satu pun task, dan d.tasks tidak memuat apa pun. Pastikan setiap task tersambung, baik melalui pemanggilan bertingkat maupun melalui penerusan objek XComArg sebagai argumen.

Menyimpan data besar pada XCom. Nilai yang dipindahkan antar task tersimpan pada basis data metadata, sehingga dataframe besar membuat dag run lambat dan ukuran basis data membengkak. Tulis data besar ke dalam berkas atau penyimpanan objek, lalu teruskan lokasi berkas atau referensinya saja.

Ringkasan#

• Decorator @task mengubah fungsi Python menjadi task, sedangkan @dag mengubah fungsi menjadi generator DAG.
• Pemanggilan fungsi task pada saat perancangan menghasilkan XComArg dan sekaligus menyimpulkan dependensi, sehingga pengkabelan XCom tidak ditulis secara manual.
• Nilai balik dict dapat dipecah menjadi beberapa XCom melalui multiple_outputs=True.
• Gaya klasik dengan PythonOperator tetap sah, tetapi menuntut xcom_pull manual dan deklarasi dependensi terpisah.

Referensi#

• TaskFlow — Apache Airflow Documentation — Mekanisme @task, XComArg, dan penyimpangan dependensi TaskFlow.
• XComs — Apache Airflow Documentation — Konsep XCom dan batas ukuran data yang dipindahkan antar task.
• Dags — Apache Airflow Documentation — Cara mendeklarasikan DAG, termasuk decorator @dag.