Belajar Data Engineer - Batch Pipeline & Airflow

#Pendahuluan

Setelah di episode 7 kalian membangun model dbt yang rapi, muncul pertanyaan operasional: siapa yang menjalankannya setiap malam? Jawabannya: orkestrator. Cron sederhana tidak cukup — pipeline produksi butuh retry otomatis, penanganan kegagalan, logging, dan visibilitas.

Episode ini membangun fondasi orkestrasi batch dengan Apache Airflow, standar industri yang (masih) paling banyak dipakai. Kalian akan memahami DAG, operators, scheduling, dan praktik membangun pipeline ekstraksi → transformasi → load yang benar-benar berjalan.

#Konsep Dasar Airflow

#DAG dan Task

Airflow bekerja dengan Directed Acyclic Graph (DAG): sekumpulan task (unit kerja) yang dihubungkan dengan ketergantungan. "Acyclic" berarti tidak boleh ada siklus — aliran selalu maju dari awal ke akhir.

Setiap task adalah satu operator: eksekusi Python, query SQL, transfer data, atau perintah Bash. DAG adalah kode Python — dan itu kekuatan Airflow: seluruh pipeline bisa di-versioning dan di-review seperti software.

#Scheduling: Logika di Balik Eksekusi

Scheduling Airflow adalah salah satu konsep yang paling sering disalahpahami. schedule_interval menentukan kapan sebuah DAG Run dimulai — dan aturan pentingnya: DAG run untuk interval T dimulai SETELAH interval T berakhir.

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime, timedelta
 
default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(hours=1),
}
 
with DAG(
    dag_id="batch_pipeline",
    default_args=default_args,
    start_date=datetime(2026, 8, 16),
    schedule_interval="0 2 * * *",   # tiap hari pukul 02:00
    catchup=False,
    max_active_runs=1,
) as dag:
    extract = PythonOperator(
        task_id="extract_orders",
        python_callable=extract_orders,   # dari episode 5
    )
    transform = PostgresOperator(
        task_id="transform_sql",
        sql="sql/clean_orders.sql",
        postgres_conn_id="de_lab",
    )
    load = PostgresOperator(
        task_id="load_marts",
        sql="sql/build_marts.sql",
        postgres_conn_id="de_lab",
    )
 
    extract >> transform >> load

Perhatikan baris terakhir: extract >> transform >> load — sintaks Airflow untuk menyusun ketergantungan. Seluruh pipeline batch kalian bisa dibangun dari pola ini.

#Menjalankan Airflow Lokal

Cara tercepat menjalankan Airflow adalah dengan Docker Compose resmi:

curl -LfO https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml
mkdir -p ./dags ./logs ./plugins ./config
docker compose up -d

Setelah berjalan, buka http://localhost:8080 (login default airflow/airflow). Airflow memakai PostgreSQL sebagai metadata database dan scheduler + webserver sebagai komponen utama — arsitektur ini kita bedah lebih dalam di episode 13.

#Monitoring Pipeline

Airflow memberikan visibilitas eksekusi secara out-of-the-box:

• Grid View: melihat status tiap task per DAG run (sukses, gagal, berjalan, retry).
• Gantt Chart: melihat berapa lama tiap task berjalan dan bottleneck-nya.
• Log per task: setiap task punya log lengkap yang bisa ditelusuri.
• Email/notification: bisa mengirim alert saat task gagal — fondasi topik monitoring di episode 20.

airflow dags list
airflow tasks test batch_pipeline extract_orders 2026-08-16
airflow dags trigger batch_pipeline

airflow tasks test menjalankan satu task saja — alat debugging terbaik untuk memisahkan "masalah pipeline" vs "masalah kode".

#Pola DAG Production-Grade

Pipeline batch produksi biasanya mengikuti pola ini:

1. Extract — ambil data dari sumber (API, database) dengan incremental logic.
2. Stage — simpan hasil ekstraksi ke staging area.
3. Transform — jalankan dbt models (dbt build).
4. Quality — jalankan test data quality.
5. Notify — kirim notifikasi sukses/gagal.

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from datetime import datetime
 
with DAG(
    dag_id="elt_daily",
    start_date=datetime(2026, 8, 16),
    schedule_interval="0 1 * * *",
    catchup=False,
) as dag:
    extract = PythonOperator(
        task_id="extract_incremental",
        python_callable=extract_incremental,   # fungsi incremental
    )
    dbt_build = BashOperator(
        task_id="dbt_build",
        bash_command="cd ~/de-lab/dbt_project && dbt build",
    )
    quality = BashOperator(
        task_id="data_quality",
        bash_command="cd ~/de-lab/dbt_project && dbt test --select severity:error",
    )
    notify = BashOperator(
        task_id="notify_success",
        bash_command="curl -s -X POST https://webhook.example/success",
    )
 
    extract >> dbt_build >> quality >> notify

Ini contoh nyata: orkestrasi memanggil tool lain (dbt) dan memberi tahu konsumen. Pipeline tidak lagi sekumpulan script, melainkan produk dengan siklus hidup — dan itulah yang membedakan data engineer junior dari senior.

#Kesalahan Umum (Common Pitfalls)

1. Salah paham scheduling. schedule_interval="0 2 * * *" berarti run untuk data kemarin mulai jam 2 pagi setelah interval — bukan jam 2 pagi untuk data hari itu. Pahami logical_date vs execution_date.

2. Menulis semua logika dalam satu task raksasa. Pecah menjadi task kecil agar retry hanya mengulang bagian yang gagal, bukan seluruh pipeline.

3. Tanpa retry dan tanpa notification. Pipeline yang gagal diam-diam adalah pembunuh kepercayaan data. Set retries dan alert.

4. Mengandalkan web UI untuk menjalankan DAG. Semua eksekusi harus lewat scheduler — trigger manual dari UI hanya untuk debugging.

#Penutup

Di episode 8 ini kalian telah membangun orkestrasi batch pipeline:

• DAG adalah pipeline sebagai kode — task, dependency, dan scheduling dalam satu file Python yang bisa di-versioning.
• Scheduling: DAG run untuk interval T dimulai setelah interval berakhir; pahami catchup dan max_active_runs.
• Monitoring: Grid View, log per task, dan notifikasi kegagalan.
• Pola produksi: extract → dbt build → quality → notify.

Di episode 9 selanjutnya kita naik ke level penyimpanan analitik: data warehouse cloud (Snowflake/BigQuery/Redshift) — arsitektur warehouse modern, partitioning & clustering, dan strategi cost management, plus praktik optimasi query di warehouse. Sampai jumpa di episode 9!