Integrasi dbt

Proyek transformasi yang tidak lagi cukup ditangani beberapa pernyataan SQL sederhana biasanya dikelola dengan dbt, yaitu alat baris perintah yang mengorganisasi transformasi SQL dalam sebuah proyek, lengkap dengan dependensi antar-model dan pengujian. Topik ini membahas dua pola integrasi dbt dengan Airflow, yaitu pemanggilan langsung melalui BashOperator dan perenderan proyek dbt menjadi task Airflow menggunakan paket Cosmos. Kedua pola dibahas beserta batas peran masing-masing alat. Setelah membacanya, pilihan pola integrasi dapat ditentukan dengan mempertimbangkan granularitas task, dependensi tambahan, dan bentuk log yang dibutuhkan.

Peran dbt pada alur transformasi#

dbt mengelola transformasi SQL di dalam basis data melalui konsep model, yaitu berkas SQL berisi pernyataan SELECT yang diberi nama dan dapat dirujuk oleh model lain. Setiap model diberi materialisasi, yaitu konfigurasi yang menentukan wujud hasilnya di basis data, misalnya tabel, view, atau tabel incremental yang hanya menambahkan baris baru. Proyek dbt juga memuat test, yaitu pernyataan pemeriksaan atas hasil model, misalnya kolom tidak boleh bernilai NULL atau nilainya harus unik. Seluruh konsep itu dieksekusi oleh perintah seperti dbt run dan dbt test pada terminal.

Yang tidak ditangani dbt adalah orkestrasi. dbt tidak memiliki penjadwal, tidak mengetahui pemuatan data dari sistem lain, dan tidak mengelola task yang berada di luar proyeknya. Penjadwalan dag run, dependensi lintas sistem, sensor atas sumber data, mekanisme retry, serta pemantauan status pada UI tetap menjadi peran Airflow. Karena itu, dbt tidak menggantikan DAG, melainkan menjalankan satu bagian transformasi di dalam DAG yang lebih besar.

Pembagian peran itu berdampak langsung pada penulisan DAG. Task dbt diletakkan setelah task pemuatan data yang menyediakan sumber datanya, dan task analitik atau pengiriman laporan diletakkan setelahnya. Urutan antar-model di dalam proyek dbt ditentukan dbt sendiri berdasarkan perintah ref antar-model, sehingga DAG tidak perlu mendaftarkan dependensinya satu per satu.

Pola BashOperator#

Pola paling ringkas adalah menjalankan dbt run melalui BashOperator, sebagaimana perintah dbt dijalankan pada terminal. Berkas dag_dbt_bash.py berikut memuat dua task yang berantai, yaitu menjalankan model lalu menguji hasilnya.

from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import dag


@dag(
    dag_id="dag_dbt_bash",
    schedule=None,
    description="Menjalankan proyek dbt lewat BashOperator.",
)
def dag_dbt_bash():
    jalankan_model = BashOperator(
        task_id="jalankan_model",
        bash_command="dbt run --project-dir /proyek-dbt",
    )

    uji_model = BashOperator(
        task_id="uji_model",
        bash_command="dbt test --project-dir /proyek-dbt",
    )

    jalankan_model >> uji_model


dag_dbt_bash()

# uji bahwa berkas DAG dapat diparse tanpa keluaran apa pun
python3 dag_dbt_bash.py

Perintah tersebut berakhir tanpa keluaran apa pun karena bash_command hanyalah string yang baru dievaluasi saat task berjalan. Eksekusi penuh DAG ini memerlukan instalasi dbt pada lingkungan eksekusi, misalnya melalui pip install dbt-core beserta adapter basis datanya. Folder /proyek-dbt juga harus memuat proyek dbt. Pada lingkungan latihan tanpa dbt, DAG tetap terdaftar dan terlihat di UI, tetapi task jalankan_model akan gagal apabila dipicu karena perintah dbt tidak ditemukan.

Kekuatan pola ini terletak pada kesederhanaannya, karena tidak ada paket tambahan apa pun di sisi Airflow. Batasnya muncul pada granularitas, yaitu satu task mewakili seluruh proyek. Kegagalan satu model membuat task jalankan_model gagal seluruhnya. Log dbt tercampur menjadi satu teks panjang di dalam log task, dan retry mengulang seluruh run alih-alih model yang gagal saja.

Profil dan proyek dbt#

Perintah dbt run membaca dua berkas konfigurasi sebelum mengakses basis data. Berkas dbt_project.yml berada di akar proyek dan menyatakan nama proyek serta folder modelnya.

# tidak-dijalankan
name: analitik_sekolah
version: "1.0.0"
profile: analitik_sekolah
model-paths: ["models"]

Berkas profiles.yml menyatakan koneksi basis data yang digunakan proyek tersebut. Secara default berkasnya berada di folder ~/.dbt, dan nilai rahasianya lazim dibaca dari variabel lingkungan agar tidak tertulis di kode.

# tidak-dijalankan
analitik_sekolah:
  target: dev
  outputs:
    dev:
      type: postgres
      host: localhost
      port: 5432
      user: pengembang
      password: "{{ env_var('DBT_PASSWORD') }}"
      dbname: analitik
      schema: public

Dengan pola BashOperator, kedua berkas itu sepenuhnya dikelola oleh dbt, dan Airflow hanya memastikan perintah dbt run dijalankan pada waktu dan urutan yang tepat.

Pola Cosmos#

Paket Cosmos, yang dipasang dengan nama astronomer-cosmos, mengambil pendekatan sebaliknya dengan merender proyek dbt menjadi DAG dan TaskGroup Airflow. Sesuai README resminya, Cosmos mengubah setiap model dbt menjadi task atau task group lengkap dengan retry dan alerting, yaitu pemberitahuan kegagalan yang terkirim secara otomatis. Cosmos juga menjalankan proyek dbt terhadap Connection Airflow alih-alih profiles.yml, serta mendukung instalasi dbt di dalam virtual environment, yaitu lingkungan Python yang terpisah, untuk menghindari konflik dependensi dengan Airflow.

# tidak-dijalankan
# memerlukan paket astronomer-cosmos beserta instalasi dbt
from airflow.sdk import dag
from cosmos import DbtTaskGroup, ProfileConfig, ProjectConfig
from cosmos.profiles import PostgresUserPasswordProfileMapping

profile_config = ProfileConfig(
    profile_name="analitik_sekolah",
    target_name="dev",
    profile_mapping=PostgresUserPasswordProfileMapping(
        conn_id="pg_local",
        profile_args={"schema": "public"},
    ),
)


@dag(dag_id="dag_dbt_cosmos", schedule=None)
def dag_dbt_cosmos():
    DbtTaskGroup(
        group_id="transformasi",
        project_config=ProjectConfig("/proyek-dbt"),
        profile_config=profile_config,
    )


dag_dbt_cosmos()

DbtTaskGroup membentuk graf task di dalam DAG, sehingga setiap model muncul sebagai simpul tersendiri pada Graph view beserta status dan lognya. ProfileConfig memetakan Connection Airflow menjadi profil dbt, sehingga kredensial cukup dikelola sekali di Airflow. Biayanya adalah dependensi baru yang versinya harus selaras dengan versi Airflow dan dbt yang digunakan. Karena itu, peningkatan versi salah satu alat perlu diperiksa terhadap matriks dukungan Cosmos.

Perbandingan kedua pola dirangkum pada tabel berikut.

Pola BashOperator cocok untuk proyek kecil atau tahap perkenalan, sedangkan Cosmos cocok ketika jumlah model membesar dan granularitas pemantauan menjadi kebutuhan.

Ringkasan#

• dbt mengelola transformasi SQL berbasis proyek melalui model, materialisasi, dan test, sedangkan penjadwalan, dependensi lintas sistem, dan sensor tetap menjadi peran Airflow.
• Pola BashOperator menjalankan dbt run dan dbt test sebagai perintah biasa, ringkas dan tanpa dependensi tambahan, tetapi satu task mewakili seluruh proyek.
• Pola Cosmos merender setiap model dbt menjadi task Airflow dengan DbtTaskGroup, menggunakan Connection sebagai sumber kredensial, beserta biaya dependensi baru yang versinya perlu diselaraskan.
• Parse tanpa keluaran hanya memastikan struktur DAG valid, sedangkan kesiapan dbt dan path proyeknya baru teruji saat task dieksekusi.

Referensi#

• Introduction — dbt Docs — Definisi dbt sebagai alat transformasi data beserta konsep model, test, dan materialisasi.
• profiles.yml — dbt Docs — Struktur berkas koneksi basis data yang dibaca oleh proyek dbt.
• Bash Operator — Apache Airflow — Panduan resmi BashOperator yang menjadi dasar pola pemanggilan langsung.
• astronomer/astronomer-cosmos — GitHub — README Cosmos yang menjelaskan perenderan proyek dbt menjadi DAG Airflow dan task per model.