Apache Airflow dan OpenLineage: Orkestrasi Pipeline Data dan Pelacakan Lineage End-to-End

Ilustrasi dihasilkan oleh AI untuk artikel ini; bukan tangkapan layar sistem nyata. Pipeline data yang berjalan tepat waktu belum tentu...

Ilustrasi dihasilkan oleh AI untuk artikel ini; bukan tangkapan layar sistem nyata.

Pipeline data yang berjalan tepat waktu belum tentu dapat dipercaya. Pertanyaan yang paling sering muncul ketika sebuah angka di dasbor tiba-tiba berubah bukanlah "apakah DAG-nya sukses", melainkan "tabel mana yang terakhir menulis ke angka ini, dari sumber apa, dan siapa yang mengubah transformasinya". Orkestrasi menjawab pertanyaan pertama; lineage menjawab sisanya. Apache Airflow dan OpenLineage adalah dua komponen yang sering dipasang berpasangan justru karena keduanya menutup celah yang berbeda: Airflow mengatur urutan eksekusi, OpenLineage merekam jejak asal-usul data yang mengalir melalui eksekusi tersebut.

Artikel ini menyusun gambaran praktis keduanya: bagaimana Airflow bekerja sebagai lapisan orkestrasi, bagaimana OpenLineage menangkap metadata dari task yang berjalan, dan bagaimana keduanya dirakit pada lab lokal dengan Docker Compose serta Marquez sebagai backend referensi. Semua konfigurasi di bawah adalah contoh lab — belum diuji end-to-end pada skala produksi, dan tidak ada klaim performa yang disertakan.

Mengapa Orkestrasi Saja Tidak Cukup

Airflow menyimpan status task, log, dan riwayat percobaan ulang. Yang tidak secara otomatis tersimpan adalah hubungan antar-data: bahwa task load_orders membaca tabel raw.orders lalu menulis ke staging.orders_clean, yang kemudian dikonsumsi tiga model dbt berbeda. Informasi itu sebenarnya ada di dalam kode dan log, tetapi tersebar, tidak terstruktur, dan cepat usang ketika operator diganti.

Akibatnya, tim data biasanya bekerja dengan peta mental. Ketika satu tabel sumber berubah skema, penelusuran dampak dilakukan manual: membuka repository, membaca ulang DAG, menanyakan ke pemilik domain. Pendekatan ini berfungsi sampai jumlah pipeline melewati ambang tertentu; setelah itu biaya investigasi naik lebih cepat daripada jumlah pipeline itu sendiri.

Tiga pertanyaan yang sulit dijawab DAG

Pertama, dampak perubahan: jika kolom customer_id di sumber berubah tipe, laporan mana yang terpengaruh. Kedua, akar masalah: ketika sebuah metrik turun, node mana pada grafik yang menjadi titik gagal pertama. Ketiga, kepatuhan dan audit: siapa atau proses apa yang menulis ke tabel tertentu pada rentang waktu tertentu, dan dengan versi kode yang mana.

Yang dibutuhkan bukan sekadar catatan tambahan, melainkan model data yang berbeda: sebuah graf berarah dengan node berupa job dan dataset, serta edge yang menyatakan "job ini membaca dataset itu" dan "job ini menulis dataset itu". Ketika metadata disimpan dalam bentuk graf, penelusuran dampak berubah menjadi traversal — dari kolom yang berubah, ikuti edge ke hilir sampai berhenti. Pertanyaan yang sebelumnya butuh satu jam rapat menjadi kueri beberapa detik, dengan syarat grafnya lengkap dan konsisten.

Ketiganya bersifat lintas-sistem. Sebuah DAG Airflow tidak tahu bahwa tabel yang ia tulis juga dipakai oleh notebook analis atau job Spark di kluster lain. Lineage yang baik harus dapat menyatukan kejadian dari beberapa orkestrator dan beberapa mesin pemrosesan ke dalam satu model grafik yang konsisten.

Airflow sebagai Lapisan Orkestrasi

Apache Airflow memodelkan pekerjaan sebagai DAG — graf asiklik berarah — yang terdiri dari task. Scheduler membaca definisi DAG dari folder dags/, memparsing berkala, lalu menempatkan task instance ke antrean sesuai dependensi dan jadwal. Webserver menyajikan antarmuka serta REST API. Metadata seperti status task, variabel koneksi, dan riwayat dijalankan disimpan pada basis data metadata, umumnya PostgreSQL untuk pemakaian serius. Executor menentukan di mana task benar-benar berjalan: LocalExecutor untuk satu host, CeleryExecutor atau KubernetesExecutor untuk beban terdistribusi.

Konsep inti yang perlu dipegang

Ada empat hal yang paling sering menentukan keberhasilan implementasi. Idempotensi: task harus aman dijalankan ulang untuk rentang tanggal yang sama, karena retry dan backfill adalah hal normal, bukan pengecualian. Interval data: setiap run memiliki data_interval_start dan data_interval_end yang menjadi kontrak pemrosesan, bukan sekadar ds yang sering disalahgunakan. Koneksi dan variabel: kredensial disimpan di metadata, tidak di dalam kode DAG. Batas tanggung jawab: Airflow mengoordinasikan, bukan memindahkan data dalam jumlah besar; transformasi berat sebaiknya didelegasikan ke mesin seperti Spark, dbt, atau warehouse itu sendiri.

Perlu dicatat bahwa istilah dan antarmuka berubah antar versi. Pada Airflow 2.x penjadwalan berbasis data memakai datasets; pada Airflow 3.x konsep itu bergeser menjadi assets dengan API yang berbeda. Provider OpenLineage mengikuti perubahan tersebut, sehingga versi Airflow, versi provider, dan backend lineage sebaiknya dipin secara eksplisit dalam berkas dependensi. Lab pada artikel ini memakai Airflow 2.10 sebagai titik acuan, dan setiap contoh kode di bawah harus disesuaikan bila pembaca memakai versi lain.

Menyiapkan Lab Lokal dengan Docker Compose

Untuk eksperimen, satu berkas Compose sudah cukup: PostgreSQL sebagai metadata database, Airflow versi terbaru dengan LocalExecutor, Marquez sebagai penerima metadata lineage, dan satu folder DAG yang di-mount dari host. Berikut kerangka yang saya pakai pada lab lokal.

# docker-compose.yml (ringkas, lab lokal)
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    volumes:
      - pgdata:/var/lib/postgresql/data

  airflow:
    image: apache/airflow:2.10.4-python3.11
    depends_on: [postgres]
    environment:
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
      AIRFLOW__CORE__LOAD_EXAMPLES: "false"
      AIRFLOW__OPENLINEAGE__TRANSPORT: '{"type":"http","url":"http://marquez:5000","endpoint":"api/v1/lineage"}'
      AIRFLOW__OPENLINEAGE__NAMESPACE: "lab-airflow"
    ports: ["8080:8080"]
    volumes:
      - ./dags:/opt/airflow/dags
    command: standalone

  marquez:
    image: marquezproject/marquez:latest
    ports: ["5000:5000", "3000:3000"]
    depends_on: [postgres]

volumes:
  pgdata:

Perintah airflow standalone menjalankan inisialisasi metadata, scheduler, dan webserver dalam satu proses — memadai untuk lab, tetapi tidak untuk produksi. Di lingkungan nyata, komponen dijalankan terpisah dan diletakkan di belakang load balancer, dengan health check pada masing-masing proses.

Catatan lab: konfigurasi di atas dirancang untuk satu mesin pengembang. Port Marquez dan Airflow terbuka tanpa autentikasi, metadata memakai kredensial default, dan tidak ada TLS. Jangan menyalin berkas ini ke lingkungan yang dapat diakses publik tanpa menambahkan jaringan internal, autentikasi, dan pembatasan akses.

Menulis DAG yang Ramah Lineage

OpenLineage bekerja paling baik ketika DAG memberi tahu apa yang dibaca dan ditulis. Untuk operator yang sudah dikenal (misalnya Postgres, S3, Spark, dbt), provider menyediakan extractor yang membaca konfigurasi task secara otomatis. Untuk PythonOperator yang memanggil fungsi arbitrer, tidak ada yang bisa ditebak — kecuali kita mendeklarasikannya lewat inlets dan outlets.

Pola task dan dataset

Pola yang saya pakai: satu task untuk ekstraksi, satu untuk transformasi, satu untuk validasi. Setiap task mendeklarasikan dataset masukan dan keluaran dengan nama yang stabil dan dapat dipetakan ke katalog, misalnya postgres://warehouse:5432/analytics.public.orders_clean. Nama yang stabil inilah yang membuat grafik lineage tetap koheren antar-run, antar-versi DAG, dan antar-orkestrator.

# dags/orders_pipeline.py
from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.datasets import Dataset
from airflow.lineage.entities import Table

RAW = Dataset("postgres://warehouse:5432/raw.public.orders")
CLEAN = Dataset("postgres://warehouse:5432/analytics.public.orders_clean")


def extract(**context):
    # contoh: baca dari sumber, tulis ke raw
    print("extract", context["data_interval_start"], context["data_interval_end"])


def transform(**context):
    print("transform", context["data_interval_start"])


with DAG(
    dag_id="orders_pipeline",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
    default_args={"retries": 2, "retry_delay": 300},
    tags=["lab", "orders"],
) as dag:
    t_extract = PythonOperator(
        task_id="extract_orders",
        python_callable=extract,
        outlets=[RAW],
    )

    t_transform = PythonOperator(
        task_id="transform_orders",
        python_callable=transform,
        inlets=[Table("postgres", "warehouse", "raw", "public", "orders")],
        outlets=[CLEAN],
    )

    t_extract >> t_transform

Ada dua jalur metadata di sini yang perlu dibedakan. Dataset dipakai Airflow untuk penjadwalan berbasis data — DAG lain dapat dipicu ketika CLEAN diperbarui. OpenLineage memakai outlets dan inlets sebagai petunjuk dataset, lalu memperkayanya dengan facet: skema kolom, statistik, kualitas data, dan informasi runtime. Keduanya saling melengkapi dan tidak saling menggantikan.

Menghubungkan OpenLineage ke Airflow

Integrasinya lewat paket apache-airflow-providers-openlineage. Provider ini mendaftarkan listener yang membungkus siklus hidup task, membangun event RunEvent bertipe START, COMPLETE, FAIL, dan ABORT, lalu mengirimkannya melalui transport yang dikonfigurasi. Event mengikuti spesifikasi OpenLineage: setiap event memiliki job, run, inputs, outputs, serta producer.

Konfigurasi emitter dan transport

Ada dua cara mengonfigurasi transport. Pertama melalui variabel lingkungan seperti pada Compose di atas. Kedua melalui berkas openlineage.yml yang dirujuk variabel OPENLINEAGE_CONFIG, lebih mudah dibaca ketika endpoint memerlukan header autentikasi atau beberapa backend sekaligus.

# openlineage.yml
transport:
  type: http
  url: http://marquez:5000
  endpoint: api/v1/lineage
  auth:
    type: api_key
    apiKey: REDACTED
  timeout: 5.0

# variabel lingkungan alternatif
export AIRFLOW__OPENLINEAGE__NAMESPACE="lab-airflow"
export AIRFLOW__OPENLINEAGE__TRANSPORT='{"type":"http","url":"http://marquez:5000","endpoint":"api/v1/lineage"}'
export OPENLINEAGE_CONFIG=/opt/airflow/openlineage.yml

# verifikasi provider terpasang
airflow providers list | grep openlineage

Beberapa hal yang sering terlewat. Namespace harus konsisten; jika berubah saat upgrade, grafik lineage terbelah menjadi dua dunia yang tidak saling terhubung. Producer sebaiknya dibiarkan mengikuti versi provider agar dapat dilacak. Transport failure tidak boleh menggagalkan task bisnis: kegagalan pengiriman metadata harus dicatat sebagai peringatan, bukan sebagai kegagalan pipeline. Dan untuk operator kustom, extractor perlu didaftarkan melalui entry point openlineage.airflow.extractors agar metadata yang relevan ikut terkirim.

Memverifikasi Lineage End-to-End

Klaim bahwa lineage "sudah jalan" hanya sah jika ada bukti. Urutan verifikasi yang saya pakai: picu satu run, periksa log task untuk event yang terkirim, lalu tanyakan ke backend lineage apakah job dan dataset benar-benar tercatat. Marquez menyediakan REST API yang memadai untuk pemeriksaan ini.

# 1. picu DAG dan tunggu selesai
airflow dags trigger orders_pipeline
airflow dags list-runs -d orders_pipeline --state success

# 2. periksa log task: cari jejak emitter OpenLineage
airflow tasks logs orders_pipeline extract_orders 2026-11-11

# 3. tanyakan ke Marquez: apakah job terdaftar?
curl -s "http://localhost:5000/api/v1/namespaces/lab-airflow/jobs" | jq '.jobs[].name'

# 4. periksa lineage dataset keluaran
curl -s "http://localhost:5000/api/v1/lineage?nodeId=dataset:postgres://warehouse:5432/analytics.public.orders_clean" | jq '.graph'

# 5. bandingkan dengan antarmuka web Marquez pada :3000

Jika langkah ketiga kosong, penyebab paling umum adalah transport yang salah tulis, namespace berbeda antara run dan kueri, atau task yang gagal sebelum event COMPLETE terkirim. Jika job muncul tetapi dataset kosong, biasanya masalahnya ada pada deklarasi inlets/outlets atau extractor yang tidak mengenali operator yang dipakai. Perlu dicatat: seluruh urutan di atas adalah rencana verifikasi pada lab lokal dan belum saya jalankan end-to-end untuk artikel ini — pembaca sebaiknya memperlakukan hasilnya sebagai hipotesis yang harus dibuktikan sendiri.

Praktik, Batasan, dan Langkah Berikutnya

Lineage bukan proyek sekali jadi. Nilainya muncul ketika cakupannya cukup luas sehingga penelusuran dampak benar-benar berhenti di satu tempat, bukan berpindah antara tiga alat. Karena itu, keputusan paling berpengaruh bukanlah pilihan backend, melainkan konvensi penamaan dataset dan disiplin mendeklarasikan masukan serta keluaran pada setiap task.

Checklist menuju produksi

Mulai dari satu domain data yang paling sering bermasalah, bukan seluruh organisasi. Terapkan konvensi nama scheme://host:port/katalog.skema.tabel dan dokumentasikan. Pasang backend lineage dengan retensi yang disepakati, lalu uji kegagalannya: matikan backend dan pastikan pipeline tetap sukses. Tambahkan pemeriksaan otomatis yang membandingkan jumlah task dengan jumlah event yang tercatat, sehingga kehilangan metadata terdeteksi sebagai anomali, bukan ditemukan berbulan-bulan kemudian. Terakhir, hubungkan lineage ke katalog data dan proses insiden, karena grafik yang tidak pernah dibuka tidak akan pernah dirawat.

Batasan yang jujur perlu disebut. Lineage yang dikumpulkan pada level task tidak otomatis memahami transformasi kolom di dalam SQL; untuk itu dibutuhkan parser SQL atau integrasi tingkat tabel seperti dbt. Lineage juga tidak menjamin kualitas data — ia menunjukkan ke mana data pergi, bukan apakah nilainya benar. Dan pada sistem dengan volume event tinggi, biaya penyimpanan serta latensi pengiriman metadata menjadi pertimbangan arsitektur tersendiri yang tidak dibahas di sini.

Dokumentasi resmi tetap menjadi rujukan utama dan lebih cepat diperbarui daripada artikel mana pun: dokumentasi Apache Airflow untuk konsep orkestrasi dan konfigurasi, dokumentasi OpenLineage untuk spesifikasi event dan facet, serta dokumentasi provider OpenLineage untuk Airflow untuk detail transport dan extractor.

Kesimpulannya sederhana: Airflow memberi tahu bahwa pekerjaan berjalan, OpenLineage memberi tahu apa yang sebenarnya terjadi pada data. Menggabungkan keduanya mengubah investigasi dari kegiatan arkeologi menjadi kueri. Untuk lab lokal, biayanya beberapa berkas konfigurasi; untuk produksi, biayanya adalah disiplin konvensi yang harus dijaga setiap hari.

Bagikan Artikel:

WhatsApp Telegram X LinkedIn Facebook