KAL
Menu

Operator Bash dan Python

Menggunakan BashOperator dan PythonOperator, dua operator paling umum pada gaya klasik penulisan DAG.

Daftar isi

BashOperator dan PythonOperator adalah dua operator yang paling sering digunakan pada gaya klasik penulisan DAG. Topik ini membahas parameter utama kedua operator, perbedaan gaya klasik dengan TaskFlow API, serta pelaksanaan DAG klasik hingga seluruh task berstatus success. Gaya klasik tetap penting untuk dikuasai karena banyak contoh, dokumentasi operator, dan kode lama masih menggunakan bentuk ini.

BashOperator#

BashOperator menjalankan satu perintah shell melalui parameter bash_command. Perintah tersebut dieksekusi di lingkungan kerja task. Status akhir task mengikuti exit code perintahnya, yaitu kode numerik yang ditinggalkan proses perintah saat berakhir. Perintah yang berakhir dengan exit code nol menandakan task success, sedangkan exit code selain nol menandai task failed, dengan pesan error yang tercatat pada log. Perilaku ini membuat BashOperator cocok untuk membungkus skrip shell yang sudah ada, misalnya skrip arsip cadangan berkas atau pembersihan folder sementara.

Parameter env menyediakan variabel lingkungan khusus untuk proses yang menjalankan perintah shell. Isinya berupa kamus, misalnya env={"KOTA": "Bandung"}, dan nilainya dirujuk oleh perintah shell sebagaimana variabel lingkungan biasa. Secara bawaan, kamus ini menggantikan lingkungan proses Airflow, sehingga variabel lain tidak terbawa ke proses perintah. Dengan append_env=True, lingkungan proses Airflow tetap diwarisi, sedangkan isi env diletakkan di atasnya. Kombinasi bash_command dan env menjadi cara ringkas mengirim konfigurasi kecil tanpa menyunting teks perintahnya.

PythonOperator#

PythonOperator memanggil fungsi Python melalui parameter python_callable. Fungsi yang dirujuk harus didefinisikan pada level modul berkas DAG, yaitu langsung di badan berkas dan bukan di dalam fungsi lain. Dengan letak seperti itu, fungsi dapat diimpor oleh proses yang menjalankan task. Argumen fungsi dikirim melalui op_args untuk argumen posisi atau op_kwargs untuk argumen kata kunci. Mekanisme kedua parameter ini identik dengan pemanggilan fungsi Python biasa, sebagaimana ditunjukkan potongan berikut.

python Contoh: padanan op_args dan op_kwargs pada pemanggilan fungsi 
def hitung_rata_rata(nilai):
    return sum(nilai) / len(nilai)

print(hitung_rata_rata(*[[70, 80, 90]]))
print(hitung_rata_rata(**{"nilai": [70, 80, 90]}))
text Keluaran 
80.0
80.0

Tanda bintang tunggal membongkar list menjadi argumen posisi, sedangkan tanda bintang ganda membongkar kamus menjadi argumen kata kunci. Operator melakukan hal yang sama saat task dieksekusi. Karena nilai merupakan satu argumen berupa list, bentuk op_args yang setara adalah op_args=[[70, 80, 90]], yaitu list yang berisi satu argumen tersebut. Fungsi dipanggil di dalam proses task, bukan saat berkas DAG diparse, sehingga logikanya tetap tertunda sampai task berjalan. Nilai kembalian fungsi tidak hilang begitu saja karena PythonOperator menyimpannya sebagai XCom, yakni mekanisme penyimpanan nilai agar dapat diambil oleh task lain.

Menyusun dan menjalankan DAG klasik#

Berkas dag_klasik.py berikut menggabungkan kedua operator dalam satu DAG dengan dua task. Task pertama mencetak pesan persiapan melalui BashOperator dengan bantuan variabel lingkungan, dan task kedua menghitung rata-rata deret nilai melalui PythonOperator dengan argumen kata kunci.

python dag_klasik.py 
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import dag


def hitung_rata_rata(nilai: list[int]) -> float:
    """Menghitung rata-rata dari deret nilai yang diterima."""
    hasil = sum(nilai) / len(nilai)
    print(f"Rata-rata: {hasil}")
    return hasil


@dag(
    dag_id="dag_klasik",
    schedule=None,
    description="Dua task gaya klasik: operator Bash dan operator Python.",
)
def dag_klasik():
    persiapan = BashOperator(
        task_id="persiapan",
        bash_command="echo Menyiapkan data nilai ujian untuk $KOTA",
        env={"KOTA": "Bandung"},
    )
    hitung = PythonOperator(
        task_id="hitung_rata_rata",
        python_callable=hitung_rata_rata,
        op_kwargs={"nilai": [70, 80, 90]},
    )

    # task hitung_rata_rata baru berjalan setelah task persiapan selesai
    persiapan >> hitung


dag_klasik()

Baris persiapan >> hitung menyatakan dependensi antar task, yaitu task hitung_rata_rata baru berjalan setelah task persiapan selesai. Pada task persiapan, teks $KOTA di dalam bash_command mengambil nilai dari kamus env, sedangkan pada task hitung_rata_rata, kamus op_kwargs diteruskan kepada fungsi sebagai argumen kata kunci.

Berkas ini disalin ke folder dags pada lingkungan latihan terpisah sebelum dijalankan.

bash 
# lingkungan latihan terpisah untuk DAG klasik
export AIRFLOW_HOME=/tmp/ra-operator-bash-dan-python
# contoh DAG bawaan dimatikan agar pemrosesan folder dags cepat
export AIRFLOW__CORE__LOAD_EXAMPLES=False
bash 
# uji bahwa berkas DAG dapat diparse tanpa keluaran apa pun
python3 dag_klasik.py
bash 
# salin berkas DAG ke folder yang dipindai Airflow
mkdir -p "$AIRFLOW_HOME/dags"
cp dag_klasik.py "$AIRFLOW_HOME/dags/"
bash 
# siapkan skema basis data metadata
# log migrasi berisi stempel waktu, sehingga tidak ditampilkan
airflow db migrate > /dev/null 2>&1

Struktur DAG yang telah dibangun dapat diperiksa sekali lagi, kali ini sekaligus menampilkan kelas operator tiap task dan hubungan dependensinya.

bash 
# cetak kelas operator tiap task beserta task turunan dari persiapan
python3 -c "import dag_klasik; d = dag_klasik.dag_klasik(); print({t.task_id: type(t).__name__ for t in d.tasks}); print(sorted(d.get_task('persiapan').downstream_task_ids))"
text Keluaran 
{'persiapan': 'BashOperator', 'hitung_rata_rata': 'PythonOperator'}
['hitung_rata_rata']

Keluaran memperlihatkan dua task dengan kelas operatornya masing-masing, dan downstream_task_ids milik task persiapan berisi hitung_rata_rata sesuai notasi panah pada berkas dag_klasik.py. Setelah struktur sesuai harapan, seluruh DAG dijalankan sekali dengan dags test.

bash 
# jalankan seluruh task DAG klasik sekali secara lokal
# log panjang dan bertanda waktu, sehingga diarahkan ke berkas
airflow dags test dag_klasik 2026-11-04 > "$AIRFLOW_HOME/dags-test.log" 2>&1

Keluaran perintahnya panjang dan bertanda waktu, sehingga diarahkan ke berkas. Pada akhir log, kedua task berstatus success dan dag run ditutup dengan status yang sama. Baris penting pada log kedua task diambil dengan pola pencarian.

Log task persiapan memuat baris Menyiapkan data nilai ujian untuk Bandung, yang membuktikan variabel lingkungan dari env benar-benar terbaca oleh perintah shell. Log task hitung_rata_rata memuat baris Rata-rata: 80.0 sesuai perhitungan atas deret 70, 80, dan 90. Nilai 80.0 itu juga tersimpan sebagai XCom hasil fungsi, dan pelacakannya dapat dilakukan pada tab XCom di UI.

bash 
# ambil baris bukti dari log kedua task
export AIRFLOW_HOME=/tmp/ra-operator-bash-dan-python
grep -o "Menyiapkan data nilai ujian untuk Bandung" "$AIRFLOW_HOME/dags-test.log"
grep -o "Rata-rata: 80.0" "$AIRFLOW_HOME/dags-test.log"
text Keluaran 
Menyiapkan data nilai ujian untuk Bandung
Rata-rata: 80.0

Perbedaan dengan TaskFlow API#

Gaya klasik memisahkan definisi fungsi dan pembuatan task, sedangkan TaskFlow API menyatukannya lewat decorator. Pada gaya klasik, pengoperan nilai antar task ditulis eksplisit melalui op_args atau op_kwargs dan hasilnya diambil dari XCom, sedangkan TaskFlow API menyambungkan nilai kembalian ke argumen task berikutnya secara otomatis. Gaya klasik tetap dipertahankan Airflow karena banyak operator pihak ketiga dan contoh lama masih menggunakannya, sehingga kemampuan membaca kedua gaya sama pentingnya dengan kemampuan menulis gaya TaskFlow.

Kesalahan umum#

  • Menulis python_callable=hitung_rata_rata() dengan tanda kurung. Fungsi langsung terpanggil saat berkas DAG diparse, sehingga parameter tersebut menerima nilai kembalian fungsi, bukan fungsinya, dan berkas gagal diparse dengan pesan bahwa python_callable harus berupa callable. Perbaikannya dengan meneruskan nama fungsi tanpa tanda kurung dan mengirim argumennya melalui op_args atau op_kwargs.
  • Menaruh logika berat pada level modul berkas DAG. Kode pada level modul dijalankan ulang setiap kali berkas diparse, sehingga pekerjaan yang seharusnya milik task ikut berulang. Perbaikannya dengan memindahkan logika ke dalam fungsi yang dirujuk python_callable.

Ringkasan#

  • BashOperator menjalankan bash_command melalui shell dan menyalin exit code perintah menjadi status task.
  • Parameter env memberi variabel lingkungan khusus pada perintah, sedangkan append_env=True membuat lingkungan proses Airflow ikut diwarisi.
  • PythonOperator memanggil python_callable dari level modul berkas DAG dengan argumen op_args atau op_kwargs, lalu menyimpan nilai kembaliannya sebagai XCom.
  • Notasi >> menghubungkan dua task sehingga task hilir berjalan setelah task hulurnya selesai, dan airflow dags test memverifikasi seluruh task secara lokal beserta lognya.

Referensi#