KAL
Menu

XCom

Mengoper data kecil antar task dengan XCom, lengkap dengan batas ukuran dan praktik terbaiknya.

Daftar isi

Task pada Airflow berjalan terisolasi satu sama lain, bahkan ketika task-task tersebut berada di mesin yang berbeda, sehingga variabel Python biasa tidak dapat dibagikan antar task. Topik ini membahas XCom sebagai mekanisme pengoper nilai kecil antar task, cara TaskFlow API menyembunyikan pengoperan nilai tersebut di balik nilai kembalian fungsi, serta batas ukuran dan praktik terbaik yang menyertainya.

Konsep XCom#

XCom, singkatan dari cross-communication, adalah mekanisme yang memungkinkan task saling bertukar nilai. Tanpa XCom, task sepenuhnya terisolasi karena masing-masing dapat berjalan pada proses dan mesin yang berbeda. Satu XCom teridentifikasi oleh kunci beserta task_id dan dag_id task asalnya, sehingga nilai dari task berbeda tidak saling tertukar.

Nilainya tidak mengalir langsung dari proses pengirim ke proses penerima. XCom tersimpan di basis data metadata Airflow, yaitu basis data yang juga menyimpan status task dan status dag run. Setiap nilai diserialisasi (serialize, diubah menjadi bentuk yang dapat disimpan lalu dibaca ulang) sebelum ditulis. Task pengirim mendorong nilainya (push), sedangkan task penerima menariknya (pull) setelah task pengirim selesai. Karena penyimpanannya berupa basis data, batas ukuran XCom menjadi keterbatasan yang nyata, dan bagian akhir topik membahas cara menanganinya.

TaskFlow API menyembunyikan pasangan push dan pull tersebut. Nilai kembalian setiap fungsi @task otomatis didorong ke XCom dengan kunci return_value, sedangkan nilai itu otomatis ditarik kembali ketika digunakan sebagai argumen task lain. Penulis DAG cukup menyusun fungsi Python biasa yang saling dipanggil, karena Airflow menyiapkan pengiriman dan penerimaan XCom secara otomatis.

DAG latihan: mengoper nilai antar dua task#

DAG berikut membentuk rantai dua task yang bertukar satu nilai. Task baca_sensor menghasilkan data suhu dari sebuah sensor, sedangkan task ringkas menerima data tersebut sebagai parameter lalu menghitung jumlah data, rata-rata, dan nilai maksimumnya. Keduanya juga mencetak hasil kerjanya ke log agar aliran nilai dapat diikuti dari keluaran eksekusi. Contoh datanya sengaja dibuat kecil, lima bilangan bulat, karena XCom dirancang untuk nilai sekecil itu.

python dag_xcom.py 
from airflow.sdk import dag, task


@dag(
    dag_id="dag_xcom",
    schedule=None,
    description="DAG latihan: mengoper nilai kecil antar task dengan XCom.",
)
def dag_xcom():
    @task
    def baca_sensor() -> list[int]:
        """Membaca data suhu lalu meneruskannya ke task berikutnya."""
        suhu = [26, 27, 25, 28, 27]
        print(f"Data yang diteruskan: {suhu}")
        return suhu

    @task
    def ringkas(suhu: list[int]) -> dict:
        """Menerima data dari task sebelumnya lalu menghitung ringkasannya."""
        hasil = {
            "jumlah_data": len(suhu),
            "rata_rata": sum(suhu) / len(suhu),
            "maksimum": max(suhu),
        }
        print(f"Ringkasan: {hasil}")
        return hasil

    ringkas(baca_sensor())


dag_xcom()

Dari sudut pandang kode, baca_sensor hanya mengembalikan list dan ringkas hanya menerima parameter list. Tidak ada pemanggilan XCom eksplisit di dalam keduanya, tetapi Airflow mendorong nilai kembalian baca_sensor ke XCom lalu menariknya kembali sebagai argumen task ringkas. Apabila rantai diperpanjang dengan task ketiga, pola pemanggilan yang sama berlaku tanpa kode penghubung tambahan. Dengan begitu, daftar parameter kedua fungsi tetap bersih dari detail infrastruktur Airflow.

Memeriksa struktur DAG dan menyiapkan lingkungan#

Pemeriksaan dimulai dari eksekusi berkas DAG yang harus berakhir tanpa keluaran apa pun. Pesan error akan tampil apabila terdapat sintaks atau impor yang keliru, sedangkan berkas yang valid selesai tanpa mencetak apa pun. Struktur yang terbentuk kemudian diperiksa melalui Python untuk mencetak dag_id beserta seluruh task_id dalam urutan alfabetis.

bash 
# uji bahwa berkas DAG dapat diparse tanpa keluaran apa pun
python3 dag_xcom.py
bash 
# muat modul DAG lalu cetak dag_id dan seluruh task_id
python3 -c "import dag_xcom; d = dag_xcom.dag_xcom(); print(d.dag_id); print(sorted(t.task_id for t in d.tasks))"
text Keluaran 
dag_xcom
['baca_sensor', 'ringkas']

Keluaran menegaskan bahwa dua task sesuai dengan nama fungsinya, dan dependensi antar keduanya baru terbentuk pada pemanggilan task di dalam fungsi DAG. Selanjutnya lingkungan latihan disiapkan. CLI Airflow membaca lokasi kerja dari variabel lingkungan AIRFLOW_HOME, yaitu folder yang memuat konfigurasi, basis data metadata, dan folder dags. Latihan pada topik ini menggunakan folder terpisah di /tmp/ra-xcom agar tidak bercampur dengan instalasi lain, sedangkan contoh DAG bawaan dimatikan supaya daftar DAG benar-benar kosong.

bash 
# lingkungan latihan terpisah agar tidak bercampur dengan instalasi lain
export AIRFLOW_HOME=/tmp/ra-xcom
# contoh DAG bawaan dimatikan supaya daftar DAG benar-benar kosong
export AIRFLOW__CORE__LOAD_EXAMPLES=False
bash 
# salin DAG ke folder dags yang dibaca CLI
mkdir -p "$AIRFLOW_HOME/dags"
cp dag_xcom.py "$AIRFLOW_HOME/dags/"

Berkas DAG disalin ke folder dags karena CLI membaca DAG dari folder tersebut, bukan dari direktori kerja. Setelah penyalinan ini, seluruh prasyarat eksekusi sudah terpenuhi, yaitu berkas yang valid, struktur yang terverifikasi, dan lingkungan yang terisolasi.

Menjalankan DAG dan mengikuti aliran nilai#

Skema basis data metadata disiapkan lebih dahulu melalui db migrate, dan log migrasinya tidak ditampilkan karena berisi stempel waktu. Perintah dags test kemudian menjalankan satu dag run secara lokal tanpa scheduler maupun worker terpisah. Tanggal yang diberikan pada argumen menjadi logical date dag run tersebut.

bash 
# siapkan skema basis data metadata
# log migrasi berisi stempel waktu, sehingga tidak ditampilkan
airflow db migrate > /dev/null 2>&1
bash 
# jalankan seluruh task DAG sekali secara lokal dengan logical date tetap
# log panjang dan bertanda waktu, sehingga diarahkan ke berkas
airflow dags test dag_xcom 2026-11-04 > "$AIRFLOW_HOME/dags-test.log" 2>&1

Keluaran perintah ini panjang dan bertanda waktu, sehingga diarahkan ke berkas. Pada bagian akhir log, kedua task berstatus success dan dag run ditutup dengan status yang sama. Status tersebut menegaskan bahwa rantai pengoperan nilai berjalan tanpa hambatan.

bash 
# ambil baris keluaran kedua task dari berkas log
export AIRFLOW_HOME=/tmp/ra-xcom
grep -o "Data yang diteruskan: \[26, 27, 25, 28, 27\]" "$AIRFLOW_HOME/dags-test.log"
grep -o "Ringkasan: {'jumlah_data': 5, 'rata_rata': 26.6, 'maksimum': 28}" "$AIRFLOW_HOME/dags-test.log"
text Keluaran 
Data yang diteruskan: [26, 27, 25, 28, 27]
Ringkasan: {'jumlah_data': 5, 'rata_rata': 26.6, 'maksimum': 28}

Aliran nilainya terbaca dari log kedua task. Task baca_sensor mencetak Data yang diteruskan: [26, 27, 25, 28, 27], lalu nilai kembaliannya didorong ke XCom dengan kunci return_value. Task ringkas menarik nilai tersebut sebagai argumen suhu dan mencetak Ringkasan: {'jumlah_data': 5, 'rata_rata': 26.6, 'maksimum': 28}. Data yang diterima ringkas sama persis dengan yang dikirim, dan nilai itu tidak dihitung ulang ataupun dibaca dari berkas.

Pada DAG nyata, mekanisme yang sama mengalirkan hasil antar task yang mungkin berjalan di mesin berbeda, misalnya task perhitungan yang meneruskan path laporan ke task unggah. XCom membuat perpindahan itu terlihat seperti pemanggilan fungsi biasa, padahal nilainya benar-benar melintasi basis data metadata.

Batas ukuran dan praktik terbaik#

Batas ukuran XCom berasal dari tempat penyimpanannya. Setiap nilai diserialisasi lalu ditulis ke basis data metadata pada setiap run, sehingga nilai kecil tidak menimbulkan masalah, sedangkan nilai besar menumpuk seiring bertambahnya jumlah run. Contoh berikut mengukur ukuran dua payload, yaitu isi data yang dibawa, dalam bentuk JSON untuk memberi gambaran skalanya.

python Contoh: mengukur ukuran payload JSON 
import json

payload_kecil = [{"suhu": 26}, {"suhu": 27}, {"suhu": 25}]
payload_besar = [{"suhu": s, "lembap": 70 + s % 10} for s in range(10_000)]

for nama, payload in (("kecil", payload_kecil), ("besar", payload_besar)):
    ukuran = len(json.dumps(payload).encode("utf-8"))
    print(f"payload {nama}: {ukuran} byte")
text Keluaran 
payload kecil: 42 byte
payload besar: 298890 byte

Tiga entri kecil hanya menempati 42 byte, sedangkan sepuluh ribu entri sudah membutuhkan hampir 300 ribu byte untuk satu nilai XCom. Angka tersebut baru mencakup ukuran serialisasinya saja, karena basis data juga menyimpan kolom pengenal seperti dag_id, task_id, dan run_id pada baris yang sama. Dokumentasi resmi menyatakan XCom hanya dirancang untuk data dalam jumlah kecil dan tidak untuk nilai besar seperti dataframe, yaitu tabel data utuh di memori.

Praktik terbaiknya sejalan dengan peringatan di atas. Untuk data besar, simpan datanya pada penyimpanan yang dirancang untuk volume seperti itu, misalnya berkas di penyimpanan bersama atau objek di object storage, lalu kirimkan hanya referensinya melalui XCom, seperti path berkas, nama partisi, atau URL. Alternatif lain adalah memasang XCom backend kustom melalui konfigurasi core.xcom_backend, termasuk backend object storage bawaan yang menyimpan nilai besar di luar basis data. Dengan pola referensi, XCom kembali menjalankan perannya, yaitu mengoper penunjuk yang kecil dan cepat dibaca, sedangkan data sesungguhnya berpindah melalui saluran yang dirancang untuk volumenya.

Ringkasan#

  • XCom mengoper nilai kecil antar task melalui basis data metadata, dan satu XCom teridentifikasi oleh kunci beserta task_id dan dag_id task asalnya.
  • TaskFlow mendorong nilai kembalian @task ke XCom dengan kunci return_value lalu menariknya otomatis saat nilai itu digunakan sebagai argumen task lain.
  • XCom hanya dirancang untuk data kecil, sehingga data besar sebaiknya disimpan di penyimpanan khusus dan hanya referensinya yang dioper, atau ditangani XCom backend kustom.

Referensi#