Belajar Data Engineer - Orkestrasi & Scheduling Lanjutan

#Pendahuluan

Di episode 8 kalian menjalankan DAG Airflow pertama. Sekarang kita taruh DAG itu di dunia nyata — dunia yang tidak selalu bersahabat: sumber data terlambat, API sedang down, partisi yang harusnya sudah ada ternyata belum muncul. Pipeline production-grade dituntut untuk menunggu dengan benar, gagal dengan benar, dan pulih dengan benar.

Episode ini membawa orkestrasi ke tingkat profesional: sensors untuk ketergantungan eksternal, retry & backfill yang tidak merusak, perbandingan objektif antara Airflow, Dagster, dan Prefect, serta pola pipeline yang benar-benar dipakai di produksi.

#Sensor: Menunggu Ketergantungan dengan Benar

Masalah klasik pipeline: DAG downstream dijalankan tapi datanya belum siap. Solusi naifnya — menunggu dengan time.sleep() — adalah pemborosan resource. Solusi yang benar: sensor, task yang menunggu kondisi tertentu tanpa mengonsumsi slot worker.

from airflow.sensors.external_task import ExternalTaskSensor
from airflow.sensors.time import TimeSensor
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta
 
with DAG(
    dag_id="consumer_pipeline",
    start_date=datetime(2026, 8, 16),
    schedule_interval="0 3 * * *",
    catchup=False,
) as dag:
    wait_for_source = ExternalTaskSensor(
        task_id="wait_for_ingest",
        external_dag_id="ingest_pipeline",
        external_task_id=None,          # menunggu seluruh DAG selesai
        timeout=1800,                    # max 30 menit menunggu
        poke_interval=60,
        mode="reschedule",               # lepas slot worker saat menunggu
    )
    process = BashOperator(
        task_id="process_data",
        bash_command="dbt build",
    )
    wait_for_source >> process

Perhatikan mode="reschedule" — saat menunggu, task melepaskan slot worker sehingga tidak membuang resource. Ini perbedaan besar antara "pipeline yang benar" dan "pipeline yang memboroskan cluster".

#Retry dan Backfill: Gagal dan Pulih dengan Benar

#Retry

Retry adalah pertahanan pertama terhadap kegagalan sementara. Aturannya: retry untuk error transien, bukan error permanen (dibahas di episode 5). Di Airflow, konfigurasi retry per-DAG atau per-task:

wait_for_source = ExternalTaskSensor(
    task_id="wait_for_ingest",
    external_dag_id="ingest_pipeline",
    retries=5,
    retry_delay=timedelta(minutes=5),
    retry_exponential_backoff=True,
    max_retry_delay=timedelta(hours=1),
)

Pola yang sama: 5 percobaan, jeda yang tumbuh eksponensial, batas maksimal 1 jam. Retry eksponensial mencegah thundering herd — banyak task yang retry bersamaan dan membebani sumber.

#Backfill

Backfill adalah menjalankan ulang DAG untuk periode yang terlewat — karena downtime, data salah, atau logika berubah. Airflow menyediakannya sebagai perintah satu baris:

airflow dags backfill \
  --start-date 2026-08-08 \
  --end-date 2026-08-14 \
  --reset-dagruns \
  batch_pipeline

Kunci aman backfill: pastikan idempotent. Task harus bisa dijalankan ulang tanpa menghasilkan data ganda — karena itulah transformasi memakai MERGE (episode 10) atau mode="overwrite" per partisi, bukan sekadar append.

#Perbandingan Orkestrator Modern

Airflow bukan satu-satunya pemain. Dagster dan Prefect tumbuh pesat karena menjawab kelemahan Airflow. Perbandingan objektif:

Perbedaan paling konseptual: Airflow berpikir dalam task, Dagster berpikir dalam data (assets). Dagster membuat aset dan dependency-nya sadar — jika sebuah tabel berubah, Dagster tahu apa yang harus di-recompute. Prefect menawarkan Python yang lebih murni dan dynamic workflows yang lebih mudah.

Pendekatan pragmatis 2026: mulai dengan Airflow karena ekosistem dan referensinya paling luas; evaluasi Dagster jika kebutuhan lineage dan asset-awareness makin besar. Jangan ganti tool hanya karena hype — ukur masalah yang ingin diselesaikan.

#Praktik: Pipeline Production-Grade

Mari rangkai semua pola menjadi DAG yang layak produksi — pipeline ELT harian dengan sensor, retry, quality check, dan notification:

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.sensors.external_task import ExternalTaskSensor
from datetime import datetime, timedelta
 
 
def send_notification(context) -> None:
    run_id = context["dag_run"].run_id
    print(f"NOTIFY: pipeline {context['dag'].dag_id} run {run_id} selesai")
 
 
default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(hours=1),
    "on_failure_callback": lambda ctx: print(f"FAIL: {ctx['task_instance_key_str']}"),
}
 
with DAG(
    dag_id="elt_production",
    default_args=default_args,
    start_date=datetime(2026, 8, 16),
    schedule_interval="0 1 * * *",
    catchup=False,
    max_active_runs=1,
) as dag:
    wait_ingest = ExternalTaskSensor(
        task_id="wait_ingest",
        external_dag_id="ingest_pipeline",
        timeout=3600,
        poke_interval=60,
        mode="reschedule",
    )
    extract = BashOperator(
        task_id="extract_api",
        bash_command="cd ~/de-lab && python extract_orders.py",
    )
    dbt_build = BashOperator(
        task_id="dbt_build",
        bash_command="cd ~/de-lab/dbt_project && dbt build --select state:modified",
    )
    quality = BashOperator(
        task_id="dbt_quality",
        bash_command="cd ~/de-lab/dbt_project && dbt test --select severity:error",
    )
    notify = PythonOperator(
        task_id="notify",
        python_callable=send_notification,
    )
 
    wait_ingest >> extract >> dbt_build >> quality >> notify

Karakteristik pipeline produksi yang terlihat di sini:

1. Menunggu dependensi (sensor) — tidak asal jalan.
2. Retry eksponensial — pulih dari kegagalan sementara.
3. Transformasi idempotent (dbt + state:modified) — aman untuk backfill.
4. Quality gate + notification — data tidak sampai konsumen tanpa lolos test, dan manusia selalu tahu hasilnya.

#Kesalahan Umum (Common Pitfalls)

1. Sleep polling, bukan sensor. time.sleep(600) membuang slot worker. Gunakan sensor dengan mode="reschedule".

2. Retry tak beraturan. Tanpa backoff, kegagalan massal menjadi thundering herd. Set retry_exponential_backoff.

3. Backfill tanpa uji idempotensi. Duplikasi data adalah bencana yang muncul lambat. Uji dua-kali-jalan.

4. Berganti orkestrator karena hype. Tool adalah alat, bukan tujuan. Pindah platform hanya jika ada masalah konkret yang terukur.

#Penutup

Di episode 13 ini kalian telah membawa orkestrasi ke level produksi:

• Sensors menunggu dependensi eksternal tanpa membuang resource — mode="reschedule" adalah kuncinya.
• Retry eksponensial untuk kegagalan sementara; backfill yang aman memerlukan idempotensi.
• Airflow vs Dagster vs Prefect: model task vs asset, pilih berdasarkan masalah, bukan hype.
• Pola production-grade: sensor → extract → dbt build → quality gate → notify.

Di episode 14 selanjutnya kita membahas kualitas: data quality & testing — freshness, volume, schema checks, dbt tests, Great Expectations, dan Soda, plus praktik membangun data quality framework. Sampai jumpa di episode 14!