KAL
Menu

Operator SQL

Menjalankan dan mentemplating SQL terhadap data warehouse dengan operator SQL.

Daftar isi

DAG yang bekerja pada ranah data jarang cukup jika hanya memindahkan berkas atau menjalankan skrip, karena banyak transformasi dikerjakan langsung di dalam basis data melalui pernyataan SQL. Topik ini membahas cara menjalankan SQL dari dalam DAG menggunakan operator SQL, mulai dari mendaftarkan koneksi ke basis data, menuliskan task SQL yang membuat tabel dan mengisinya, hingga mentemplating pernyataan agar filternya berubah setiap dag run. Contoh sepanjang topik menggunakan koneksi sqlite agar seluruh langkah dapat dijalankan di lingkungan lokal, sedangkan polanya berlaku sama untuk basis data apa pun yang didukung Airflow.

Operator SQLExecuteQueryOperator#

SQLExecuteQueryOperator adalah operator dari paket provider common.sql, yaitu paket terpisah yang menambahkan dukungan SQL generik ke Airflow. Operator ini menjalankan pernyataan SQL pada koneksi apa pun yang didukung provider yang terpasang, sehingga satu cara penulisan dapat digunakan untuk sqlite, Postgres, atau data warehouse seperti DuckDB tanpa mengganti kelas operator. Parameter utamanya antara lain conn_id yang menunjuk koneksi terdaftar, sql berupa satu string atau daftar string pernyataan, serta handler yang mengatur pengambilan hasil query. Parameter autocommit menentukan apakah tiap pernyataan langsung dikomit, sedangkan split_statements memecah satu string menjadi beberapa pernyataan.

Sebagaimana operator lain, field yang dapat ditemplatkan dinyatakan pada atribut template_fields. Daftar field tersebut dapat ditampilkan langsung dari kelas operator.

python 
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

print(SQLExecuteQueryOperator.template_fields)
text Keluaran 
('sql', 'parameters', 'conn_id', 'database', 'hook_params')

Field sql berada di dalam daftar tersebut. Karena itu, pernyataan SQL boleh memuat ekspresi Jinja yang dirender setiap kali task berjalan. Field parameters dan database juga ikut ditemplatkan, sedangkan field lain seperti autocommit dikirim apa adanya.

Menyiapkan koneksi sqlite#

Operator SQL tidak menyimpan kredensial di dalam kode, melainkan merujuk koneksi terdaftar melalui conn_id. Koneksi dikelola melalui UI atau CLI, dan latihan ini menggunakan CLI agar setiap langkah dapat diulang dari terminal. Basis data latihan berupa satu berkas sqlite di /tmp/ra-operator-sql/nilai.db, dan tipe sqlite terdaftar karena paket provider sqlite sudah terpasang.

bash 
# lingkungan latihan terpisah agar tidak bercampur dengan instalasi lain
export AIRFLOW_HOME=/tmp/ra-operator-sql
export AIRFLOW__CORE__LOAD_EXAMPLES=False
bash 
# siapkan skema basis data metadata
# log migrasi berisi stempel waktu, sehingga tidak ditampilkan
airflow db migrate > /dev/null 2>&1
bash 
# daftarkan koneksi menuju berkas sqlite
# koneksi lama dihapus agar pendaftaran selalu bersih dan dapat diulang
airflow connections delete nilai_sqlite > /dev/null 2>&1 || true
airflow connections add nilai_sqlite --conn-type sqlite --conn-host /tmp/ra-operator-sql/nilai.db > /dev/null 2>&1

Pada perintah terakhir, --conn-host diisi lokasi berkas basis data karena hook sqlite membaca field host sebagai path berkas. Hook sendiri merupakan kelas pembungkus koneksi basis data milik provider. Rincian koneksi yang baru dibuat dapat diperiksa dengan airflow connections get nilai_sqlite, dan hasilnya memperlihatkan tipe sqlite beserta path pada field host. Hook juga membuat berkas basis data secara otomatis saat koneksi dibuka pertama kali, sehingga tidak ada langkah pembuatan berkas yang terpisah.

Menjalankan SQL dari DAG#

Berkas dag_sql.py berikut memuat dua task SQL yang berantai. Task pertama membuat tabel, mengosongkan isinya agar DAG dapat diulang, lalu mengisi lima baris data nilai ujian. Task kedua menjalankan query agregasi dengan filter yang ditemplatkan dari params.

python dag_sql.py 
from airflow.providers.common.sql.hooks.handlers import fetch_all_handler
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.sdk import dag


@dag(
    dag_id="dag_sql",
    schedule=None,
    description="DAG latihan: menjalankan SQL pada koneksi sqlite.",
    params={"batas": 70},
)
def dag_sql():
    siapkan_tabel = SQLExecuteQueryOperator(
        task_id="siapkan_tabel",
        conn_id="nilai_sqlite",
        sql=[
            "CREATE TABLE IF NOT EXISTS nilai_ujian (murid TEXT, kelas TEXT, nilai INTEGER)",
            "DELETE FROM nilai_ujian",
            (
                "INSERT INTO nilai_ujian VALUES "
                "('Ali', 'A', 82), ('Budi', 'A', 74), ('Citra', 'B', 91), "
                "('Dewi', 'B', 66), ('Eka', 'A', 78)"
            ),
        ],
        split_statements=True,
        autocommit=True,
    )

    ringkas_rata_rata = SQLExecuteQueryOperator(
        task_id="ringkas_rata_rata",
        conn_id="nilai_sqlite",
        sql=(
            "SELECT kelas, COUNT(*), ROUND(AVG(nilai), 1) "
            "FROM nilai_ujian WHERE nilai >= {{ params.batas }} "
            "GROUP BY kelas ORDER BY kelas"
        ),
        handler=fetch_all_handler,
        show_return_value_in_logs=True,
    )

    siapkan_tabel >> ringkas_rata_rata


dag_sql()

Field sql menerima daftar string, dan setiap elemen dieksekusi sebagai satu pernyataan secara berurutan. split_statements=True diaktifkan karena task pertama mengirim beberapa pernyataan, sedangkan autocommit=True membuat setiap pernyataan langsung dikomit. Pada task kedua, handler=fetch_all_handler menyebabkan baris hasil query diambil dan disimpan ke XCom, sedangkan show_return_value_in_logs=True mencetak nilai kembalian itu ke log task agar terbaca tanpa membuka UI.

bash 
# uji bahwa berkas DAG dapat diparse tanpa keluaran apa pun
python3 dag_sql.py
bash 
# salin berkas DAG ke folder yang dipindai Airflow
mkdir -p "$AIRFLOW_HOME/dags"
cp dag_sql.py "$AIRFLOW_HOME/dags/"
bash 
# jalankan seluruh task DAG sekali secara lokal
# log panjang dan bertanda waktu, sehingga diarahkan ke berkas
airflow dags test dag_sql 2026-11-04 > "$AIRFLOW_HOME/dags-test.log" 2>&1
grep -oE "Operator output is: \[\[.*\]\]" "$AIRFLOW_HOME/dags-test.log"
text Keluaran 
Operator output is: [[('A', 3, 78.0), ('B', 1, 91.0)]]

Log dari perintah tersebut panjang dan bertanda waktu, sehingga diarahkan ke berkas dan hanya baris nilai kembalian yang ditampilkan. Baris tersebut berasal dari task ringkas_rata_rata. Daftar luarnya mewakili satu pernyataan, sedangkan daftar di dalamnya mewakili baris hasil, yaitu kelas A dihitung dari tiga nilai dengan rata-rata 78.0 dan kelas B dari satu nilai 91.0. Nilai 66 milik Dewi tidak ikut dihitung karena berada di bawah batas params.batas, yaitu 70. Pada akhir log, kedua task berstatus success dan dag run ditutup dengan status yang sama.

Templating Jinja pada SQL#

Ekspresi {{ params.batas }} dirender tepat sebelum query dieksekusi, sehingga SQL yang diterima basis data berisi angka 70, bukan teks ekspresinya. Parameter params dideklarasikan pada decorator @dag sehingga nilainya milik DAG dan dapat ditimpa per dag run melalui UI atau CLI tanpa menyunting kode. Pola ini berguna ketika DAG yang sama harus dijalankan untuk batas filter yang berbeda, misalnya ringkasan nilai dengan ambang yang naik setiap semester.

Field parameters menawarkan cara kedua yang lebih aman untuk mengirim nilai, yaitu bind parameter yang disediakan DBAPI, standar antarmuka Python menuju basis data. Bentuknya seperti sql="SELECT * FROM nilai_ujian WHERE nilai > %s", parameters=(70,) pada hook yang mendukung fitur ini. Nilai parameter dikirim terpisah dari teks SQL, sehingga tidak dapat mengubah struktur pernyataan. Templating Jinja dipilih ketika bagian yang berubah berupa struktur pernyataan, seperti nama tabel atau klausa, sedangkan bind parameter dipilih untuk nilai yang murni data.

Transformasi agregasi dengan SQL#

Kedua contoh berikut menampilkan pola agregasi yang sama dengan DAG latihan pada bagian sebelumnya, tetapi dijalankan di luar DAG agar mekanisme SQL-nya terlihat jelas. Blok pertama memfilter nilai di atas ambang, lalu mengelompokkan hasilnya per kelas.

sql 
CREATE TABLE nilai_ujian (murid TEXT, kelas TEXT, nilai INTEGER);
INSERT INTO nilai_ujian VALUES
  ('Ali', 'A', 82), ('Budi', 'A', 74), ('Citra', 'B', 91),
  ('Dewi', 'B', 66), ('Eka', 'A', 78);
SELECT kelas, COUNT(*) AS jumlah, ROUND(AVG(nilai), 1) AS rata_rata
FROM nilai_ujian
WHERE nilai >= 70
GROUP BY kelas
ORDER BY kelas;
text Keluaran 
A|3|78.0
B|1|91.0

Blok kedua menuliskan hasil agregasi ke tabel baru menggunakan CREATE TABLE AS, yaitu pola yang lazim digunakan di data warehouse karena transformasi berjalan di dalam basis data, bukan di dalam kode task.

sql 
CREATE TABLE nilai_ujian (murid TEXT, kelas TEXT, nilai INTEGER);
INSERT INTO nilai_ujian VALUES
  ('Ali', 'A', 82), ('Budi', 'A', 74), ('Citra', 'B', 91),
  ('Dewi', 'B', 66), ('Eka', 'A', 78);
CREATE TABLE ringkasan_kelas AS
SELECT kelas, COUNT(*) AS jumlah, ROUND(AVG(nilai), 1) AS rata_rata
FROM nilai_ujian
GROUP BY kelas;
SELECT * FROM ringkasan_kelas ORDER BY kelas;
text Keluaran 
A|3|78.0
B|2|78.5

Hasilnya berbeda dari blok pertama karena tidak ada filter, sehingga kelas B menghitung dua nilai dengan rata-rata 78.5. Pada DAG sungguhan, blok semacam ini dipindahkan apa adanya ke field sql. Tabel hasil agregasi kemudian menjadi sumber bagi task atau tabel di hilir.

Kesalahan umum#

  • Mengisi conn_id yang belum terdaftar. Task berjalan, lalu gagal dengan pesan error bahwa koneksi tidak ditemukan karena pencarian koneksi terjadi saat eksekusi, bukan saat berkas DAG diparse. Perbaikannya dengan mendaftarkan koneksi melalui CLI atau UI pada AIRFLOW_HOME yang sama dengan lingkungan eksekusi DAG.
  • Menggabungkan beberapa pernyataan dalam satu string tanpa split_statements=True. Basis data menerima seluruh teks sebagai satu pernyataan dan menolaknya dengan pesan error sintaks. Perbaikannya dengan mengirim daftar pernyataan pada sql atau mengaktifkan pembagian pernyataan.
  • Menjalankan query hasil tanpa handler. Operator mengeksekusi query, tetapi baris hasilnya tidak diambil sehingga XCom dan log tidak memuat apa pun. Perbaikannya dengan mengisi handler=fetch_all_handler atau handler lain sesuai kebutuhan.

Ringkasan#

  • SQLExecuteQueryOperator dari provider common.sql menjalankan SQL pada koneksi apa pun yang didukung provider yang terpasang, dengan conn_id, sql, split_statements, dan autocommit sebagai parameter inti.
  • Koneksi didaftarkan melalui CLI atau UI dan dirujuk melalui conn_id, sehingga kredensial tidak tertulis di kode DAG.
  • Field sql ditemplatkan, sehingga filter dan bagian lain pernyataan dapat berubah per dag run melalui params atau konteks lain.
  • Hasil query hanya diambil dan didorong ke XCom apabila handler diisi, dan show_return_value_in_logs=True menampilkannya di log task.

Referensi#