KAL
Menu

Hook basis data

Menyambung ke Postgres dan S3 secara bersih melalui hook beserta manajemen koneksinya.

Daftar isi

Task yang membaca atau menulis data pada layanan eksternal memerlukan koneksi ke layanan tersebut, dan kredensial yang tertulis langsung di dalam kode DAG ikut tersalin ke repository setiap kali DAG dikirim. Topik ini membahas hook, yaitu kelas pembungkus antarmuka ke layanan eksternal yang disediakan paket provider, beserta manajemen koneksi yang menjadi dasar kerjanya. Pembahasan mencakup cara hook mengambil kredensial dari Connection serta cara mendaftarkan koneksi melalui antarmuka web, CLI, dan variabel lingkungan. Contoh lengkapnya dijalankan menggunakan SqliteHook, dan pola penggunaan PostgresHook serta S3Hook juga diuraikan.

Hook sebagai abstraksi koneksi#

Hook adalah kelas pada paket provider yang menyediakan antarmuka menuju satu jenis layanan eksternal, misalnya basis data, penyimpanan objek, atau API HTTP. Paket provider yang dipasang bersama Airflow mendaftarkan hook masing-masing, sehingga setiap jenis koneksi yang didukung memiliki kelas pembungkusnya sendiri. Kelas induk semua hook bernama BaseHook, dan setiap hook menyimpan nama argumen conn_id miliknya, misalnya sqlite_conn_id pada SqliteHook dan postgres_conn_id pada PostgresHook. Metode utama hook adalah get_conn(). Metode ini membaca koneksi terdaftar melalui conn_id, menyusun parameter sambungan dari field pada Connection, lalu mengembalikan objek koneksi milik driver, yaitu pustaka yang berkomunikasi langsung dengan basis data.

Satu panggilan get_conn() membuka satu koneksi baru, sehingga task yang memanggilnya bertanggung jawab menutup objek tersebut setelah selesai. Sebagian hook juga menyediakan metode praktis seperti get_records() yang membuka koneksi, menjalankan query, dan menutupnya dalam satu langkah. Dokumentasi resmi menganjurkan hook dibuat di dalam fungsi task karena berkas DAG dipindai berkali-kali, sedangkan isi task hanya berjalan saat eksekusi.

Dibandingkan dengan koneksi mentah yang dibuka langsung di dalam task, hook memberikan dua keunggulan. Kredensial tersimpan pada Connection, sehingga kode DAG hanya menyebut conn_id tanpa pernah berisi password. Selain itu, hook yang sama dapat digunakan banyak task dan banyak DAG, dan operator pun mengandalkan hook secara internal, misalnya SQLExecuteQueryOperator yang menjalankan SQL melalui hook untuk koneksi terdaftar.

Manajemen koneksi#

Connection menyimpan kumpulan field kredensial untuk satu layanan eksternal. Field utamanya adalah conn_id sebagai pengenal unik dan conn_type sebagai jenis layanannya, disertai field host, schema, login, password, port, serta extra berupa JSON untuk parameter tambahan. Airflow menyediakan tiga cara mengelolanya, yaitu melalui antarmuka web, melalui perintah CLI seperti airflow connections add dan airflow connections get, serta melalui variabel lingkungan bernama AIRFLOW_CONN_<CONN_ID> yang berisi URI koneksi. Nilai conn_type menentukan kelas hook yang melayani koneksi tersebut, sehingga koneksi bertipe sqlite dilayani SqliteHook dan koneksi bertipe postgres dilayani PostgresHook.

Latihan pada topik ini menggunakan koneksi sqlite yang mengarah ke berkas basis data di /tmp/ra-hook-basis-data/nilai.db. Pada hook sqlite, field host berisi path berkas basis data alih-alih nama server. Lingkungan latihannya disiapkan pada folder terpisah, skema basis data metadata dimigrasikan sekali, lalu koneksinya didaftarkan melalui CLI. Koneksi lama dihapus terlebih dahulu agar urutan perintah dapat dijalankan ulang. Keluaran perintah pendaftaran bercampur dengan log berstempel waktu, sehingga yang ditampilkan hanya tabel konfirmasinya mulai dari baris header.

bash 
# lingkungan latihan terpisah agar DAG bawaan tidak ikut dipindai
export AIRFLOW_HOME=/tmp/ra-hook-basis-data
export AIRFLOW__CORE__LOAD_EXAMPLES=False
bash 
# siapkan skema basis data metadata
# log migrasi berisi stempel waktu, sehingga tidak ditampilkan
export AIRFLOW_HOME=/tmp/ra-hook-basis-data
airflow db migrate > /dev/null 2>&1
bash 
# daftarkan koneksi menuju berkas basis data latihan
# koneksi lama dihapus agar perintah dapat dijalankan ulang
# log berstempel waktu dikecualikan dengan menampilkan tabel konfirmasinya saja
export AIRFLOW_HOME=/tmp/ra-hook-basis-data
airflow connections delete sqlite_default > /dev/null 2>&1 || true
airflow connections add sqlite_default --conn-type sqlite --conn-host /tmp/ra-hook-basis-data/nilai.db | grep -A 2 "^conn_id"
text Keluaran 
conn_id        | conn_type | host                             | login | port | extra
===============+===========+==================================+=======+======+======
sqlite_default | sqlite    | /tmp/ra-hook-basis-data/nilai.db | None  | None | {}

Konfirmasinya berupa tabel yang memuat field koneksi yang tersimpan, dengan conn_type bernilai sqlite dan field host berisi path berkas basis data latihan. Setelah koneksi terdaftar, kode DAG cukup menyebut pengenal sqlite_default, dan seluruh rincian sambungannya tersimpan pada Connection.

Membaca tabel nilai melalui SqliteHook#

Basis data latihan yang dibaca melalui SqliteHook dibuat oleh berkas isi_db.py, yaitu skrip yang membuat berkas sqlite berisi tabel nilai dengan lima baris nilai ujian. Penghapusan tabel di awal skrip membuatnya dapat dijalankan ulang tanpa menduplikasi baris, dan skrip ini dijalankan sekali sebelum DAG dieksekusi. Skrip berakhir tanpa keluaran apa pun, dan isinya terbukti benar ketika DAG membaca basis data tersebut pada langkah berikutnya.

python isi_db.py 
import sqlite3
from pathlib import Path

PATH_DB = Path("/tmp/ra-hook-basis-data/nilai.db")
NILAI = [("Ali", 82), ("Budi", 74), ("Citra", 91), ("Dewi", 66), ("Eka", 78)]

PATH_DB.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(PATH_DB)
try:
    conn.execute("DROP TABLE IF EXISTS nilai")
    conn.execute("CREATE TABLE nilai (siswa TEXT, nilai INTEGER)")
    conn.executemany("INSERT INTO nilai VALUES (?, ?)", NILAI)
    conn.commit()
finally:
    conn.close()
bash 
# buat basis data latihan beserta tabel dan barisnya
# skrip berakhir tanpa keluaran apa pun
python3 isi_db.py

Berkas dag_sqlite.py memuat satu DAG bernama dag_sqlite dengan satu task yang membaca dan merangkum tabel nilai melalui SqliteHook. Hook dibuat di dalam fungsi task dengan sqlite_conn_id="sqlite_default", yaitu pengenal koneksi yang didaftarkan pada bagian sebelumnya. Pola pertama menggunakan get_records(), yang menjalankan query dan mengembalikan seluruh barisnya tanpa penanganan koneksi manual. Pola kedua menggunakan get_conn(), yang mengembalikan objek koneksi DBAPI, yaitu objek yang mengikuti standar antarmuka Python menuju basis data dan dikelola sendiri oleh task. Pada pola kedua, task membuat cursor, yaitu objek untuk menelusuri baris hasil query, menjalankan query, mengambil hasilnya melalui fetchone(), lalu menutup koneksi pada blok finally agar tetap tertutup ketika query error.

python dag_sqlite.py 
from airflow.providers.sqlite.hooks.sqlite import SqliteHook
from airflow.sdk import dag, task


@dag(
    dag_id="dag_sqlite",
    schedule=None,
    description="Membaca dan merangkum tabel nilai melalui SqliteHook.",
)
def dag_sqlite():
    @task
    def rangkum_nilai():
        hook = SqliteHook(sqlite_conn_id="sqlite_default")

        # get_records membuka koneksi, menjalankan query, lalu menutupnya
        baris = hook.get_records("SELECT siswa, nilai FROM nilai ORDER BY siswa")
        for siswa, nilai in baris:
            print(f"{siswa}: {nilai}")

        # get_conn memberi koneksi DBAPI yang ditutup sendiri oleh task
        conn = hook.get_conn()
        try:
            cursor = conn.cursor()
            cursor.execute("SELECT ROUND(AVG(nilai), 2) FROM nilai")
            rata_rata = cursor.fetchone()[0]
            cursor.close()
        finally:
            conn.close()
        print(f"rata-rata nilai: {rata_rata}")
        return rata_rata

    rangkum_nilai()


dag_sqlite()

Langkah berikutnya dijalankan berurutan. Berkas DAG diparse terlebih dahulu sebagai pemeriksaan, disalin ke folder dags, lalu dijalankan melalui airflow dags test. Log perintahnya panjang dan bertanda waktu, sehingga diarahkan ke berkas.

bash 
# uji bahwa berkas DAG dapat diparse tanpa keluaran apa pun
export AIRFLOW_HOME=/tmp/ra-hook-basis-data
python3 dag_sqlite.py
bash 
# salin berkas DAG ke folder yang dipindai Airflow
export AIRFLOW_HOME=/tmp/ra-hook-basis-data
mkdir -p "$AIRFLOW_HOME/dags"
cp dag_sqlite.py "$AIRFLOW_HOME/dags/"
bash 
# jalankan seluruh task DAG sekali secara lokal
# log panjang dan bertanda waktu, sehingga diarahkan ke berkas
export AIRFLOW_HOME=/tmp/ra-hook-basis-data
airflow dags test dag_sqlite 2026-11-03 > "$AIRFLOW_HOME/dags-test.log" 2>&1
bash 
# tampilkan pernyataan SQL hook dan baris keluaran task dari berkas log
export AIRFLOW_HOME=/tmp/ra-hook-basis-data
grep -oE "Running statement: SELECT siswa, nilai FROM nilai ORDER BY siswa, parameters: None" "$AIRFLOW_HOME/dags-test.log"
grep -oE "(Ali|Budi|Citra|Dewi|Eka): [0-9]+|rata-rata nilai: [0-9.]+" "$AIRFLOW_HOME/dags-test.log"
text Keluaran 
Running statement: SELECT siswa, nilai FROM nilai ORDER BY siswa, parameters: None
Ali: 82
Budi: 74
Citra: 91
Dewi: 66
Eka: 78
rata-rata nilai: 78.2

Keluaran task memperlihatkan mekanisme hook secara nyata. Baris Running statement berasal dari logger milik SqliteHook yang mencatat pernyataan SQL sebelum dijalankan, sehingga kelas hook yang melayani koneksi terbaca langsung dari log. Lima baris nilai ditampilkan terurut menurut nama siswa, dan rata-rata 78.2 diperoleh melalui koneksi kedua pada task yang sama. Log juga menunjukkan task berakhir success dan dag run ditandai successful, yang berarti koneksi terbaca dari Connection terdaftar tanpa satu pun kredensial pada kode DAG.

Hook untuk Postgres dan S3#

PostgresHook melayani koneksi bertipe postgres dan termasuk keluarga yang sama dengan SqliteHook, yaitu turunan kelas DbApiHook dari provider common.sql. Karena itu, metodenya sama, yaitu get_conn(), get_records(), get_first(), run(), dan insert_rows(). Argumen koneksi bernama postgres_conn_id secara bawaan menunjuk koneksi bernama postgres_default. Hook ini tersedia pada paket apache-airflow-providers-postgres, dan pola penggunaannya identik dengan contoh sqlite pada bagian sebelumnya.

python Contoh: penggunaan PostgresHook 
# tidak-dijalankan
from airflow.providers.postgres.hooks.postgres import PostgresHook

hook = PostgresHook(postgres_conn_id="postgres_default")
baris = hook.get_records("SELECT kelas, COUNT(*) FROM nilai_ujian GROUP BY kelas")
hook.insert_rows(table="rekap_kelas", rows=[("A", 3), ("B", 2)])

S3Hook dari paket apache-airflow-providers-amazon melayani penyimpanan objek S3 dan tidak berbasis DBAPI, karena antarmukanya berupa operasi objek dan bukan query. Argumennya bernama aws_conn_id dan secara bawaan menunjuk koneksi aws_default, sedangkan kredensial AWS tersimpan pada field koneksi tersebut. Metode yang sering digunakan antara lain check_for_key() untuk memeriksa keberadaan objek, read_key() untuk membaca isi objek, load_string() untuk menulis string sebagai objek, serta get_key() yang mengembalikan kunci beserta metadatanya.

python Contoh: penggunaan S3Hook 
# tidak-dijalankan
from airflow.providers.amazon.aws.hooks.s3 import S3Hook

hook = S3Hook(aws_conn_id="aws_default")
hook.load_string("arsip nilai ujian", key="nilai/2026-11-03.csv", bucket_name="arsip-nilai", replace=True)
isi = hook.read_key(key="nilai/2026-11-03.csv", bucket_name="arsip-nilai")

Kedua contoh di atas tidak dijalankan pada lingkungan latihan ini karena memerlukan layanan PostgreSQL dan S3 yang belum tersedia. Meski begitu, polanya sama dengan SqliteHook, yaitu membuat hook dengan conn_id, memanggil metodenya di dalam task, dan membiarkan kredensial tersimpan pada Connection.

Kesalahan umum#

  • Mengimpor hook dari provider yang belum terpasang. Berkas DAG gagal diparse dengan pesan ModuleNotFoundError, sehingga DAG tidak muncul sama sekali. Perbaikannya dengan memasang paket provider yang sesuai, misalnya pip install apache-airflow-providers-postgres, pada lingkungan tempat scheduler dan worker berjalan.
  • Mengisi conn_id yang belum terdaftar. Pencarian koneksi terjadi saat task berjalan, bukan saat berkas DAG diparse, sehingga errornya baru muncul di tengah eksekusi. Perbaikannya dengan mendaftarkan koneksi pada AIRFLOW_HOME yang sama dengan lingkungan eksekusi dan menyamakan penulisan conn_id-nya.
  • Membuat hook pada tingkat modul berkas DAG. Hook mencari koneksinya setiap kali berkas dipindai, sehingga parser mengakses basis data metadata berulang kali tanpa keperluan. Perbaikannya dengan membuat hook di dalam fungsi task, sebagaimana contoh pada topik ini.

Ringkasan#

  • Hook menyediakan antarmuka ke layanan eksternal, membaca kredensial dari Connection melalui conn_id, dan mengembalikan objek koneksi yang siap digunakan melalui get_conn().
  • Connection dikelola melalui antarmuka web, CLI, atau variabel lingkungan AIRFLOW_CONN_<CONN_ID>, dan conn_type pada koneksi menentukan kelas hook yang melayaninya.
  • SqliteHook membaca field host sebagai path berkas basis data, dan get_records() serta get_conn() memberi dua tingkat kendali koneksi yang sama-sama ditutup setelah digunakan.
  • PostgresHook mengikuti antarmuka DbApiHook yang sama dengan SqliteHook, sedangkan S3Hook mengekspos operasi objek seperti read_key() dan load_string() dengan kredensial pada koneksi aws_default.

Referensi#