Pendahuluan

PostgreSQL jauh lebih dari sekadar database relasional biasa. Dengan fungsi analitik canggih, tipe data fleksibel, dan kemampuan ekstensi yang luas, PostgreSQL menjadi tulang punggung yang kuat untuk alur kerja machine learning. Banyak engineer ML yang meremehkan apa yang bisa dilakukan langsung di lapisan database, seringkali menarik data mentah ke Python untuk transformasi yang sebenarnya bisa ditangani PostgreSQL dengan lebih efisien.

Tutorial ini membahas teknik-teknik PostgreSQL lanjutan yang langsung dapat diterapkan pada pipeline ML: rekayasa fitur dengan window functions, pengorganisasian query dengan CTE, caching fitur terhitung dengan materialized views, penyimpanan metadata eksperimen dengan JSONB, penanganan dataset besar dengan partisi, pemantauan performa query, dan integrasi semuanya dengan Python menggunakan psycopg2 dan SQLAlchemy.

Prasyarat

• PostgreSQL 14+ terinstal dan berjalan
• Python 3.9+ dengan pip
• Pengetahuan dasar SQL (SELECT, JOIN, GROUP BY)
• Pemahaman konsep ML (fitur, data pelatihan, eksperimen)
• Instal paket Python yang dibutuhkan:

# Instal dependensi
pip install psycopg2-binary sqlalchemy pandas

import psycopg2
import sqlalchemy
import pandas as pd

Window Functions untuk Rekayasa Fitur

Window functions adalah salah satu alat paling ampuh untuk rekayasa fitur ML. Fungsi ini memungkinkan Anda menghitung agregat pada baris-baris yang berhubungan dengan baris saat ini tanpa menciutkan hasil — sangat cocok untuk membuat statistik bergulir (rolling statistics), fitur lag, dan fitur peringkat.

Statistik Bergulir (Rolling Statistics)

-- Buat tabel transaksi contoh
CREATE TABLE transactions (
    id SERIAL PRIMARY KEY,
    customerid INTEGER NOT NULL,

    amount DECIMAL(10, 2) NOT NULL,
    category VARCHAR(50),
    createdat TIMESTAMP NOT NULL DEFAULT NOW()
);

-- Rata-rata transaksi bergulir (7 hari terakhir) per pelanggan
SELECT
    id,
    customerid,

    amount,
    createdat,
    AVG(amount) OVER (
        PARTITION BY customerid

        ORDER BY createdat
        RANGE BETWEEN INTERVAL '7 days' PRECEDING AND CURRENT ROW
    ) AS rata2bergulir7h,
    STDDEV(amount) OVER (
        PARTITION BY customerid

        ORDER BY createdat
        RANGE BETWEEN INTERVAL '7 days' PRECEDING AND CURRENT ROW
    ) AS stddevbergulir7h,
    COUNT() OVER (

        PARTITION BY customerid

        ORDER BY createdat
        RANGE BETWEEN INTERVAL '7 days' PRECEDING AND CURRENT ROW
    ) AS jumlahtxn7h
FROM transactions
ORDER BY customerid, createdat;

-- Fitur lag: jumlah transaksi sebelumnya dan delta waktu
SELECT
    id,
    customerid,

    amount,
    createdat,
    LAG(amount, 1) OVER w AS jumlahsebelumnya1,
    LAG(amount, 2) OVER w AS jumlahsebelumnya2,
    LAG(amount, 3) OVER w AS jumlahsebelumnya3,
    amount - LAG(amount, 1) OVER w AS selisihjumlah,

    EXTRACT(EPOCH FROM (
        createdat - LAG(createdat, 1) OVER w

    )) / 3600.0 AS jamsejaktxnterakhir,
    NTILE(10) OVER (
        PARTITION BY customerid

        ORDER BY amount
    ) AS desiljumlah
FROM transactions
WINDOW w AS (PARTITION BY customerid ORDER BY createdat);

-- Peringkat pelanggan berdasarkan pengeluaran di setiap kategori
SELECT
    customerid,

    category,
    SUM(amount) AS totalbelanja,
    RANK() OVER (
        PARTITION BY category
        ORDER BY SUM(amount) DESC
    ) AS peringkatbelanja,

    PERCENTRANK() OVER (
        PARTITION BY category
        ORDER BY SUM(amount) DESC
    ) AS persentilbelanja

FROM transactions
GROUP BY customerid, category;

-- Pipeline rekayasa fitur lengkap menggunakan CTE
WITH statistikpelanggan AS (

    -- Langkah 1: Agregasi dasar per pelanggan
    SELECT
        customerid,
        COUNT() AS totaltransaksi,

        SUM(amount) AS totalbelanja,
        AVG(amount) AS rata2jumlah,

        STDDEV(amount) AS stddevjumlah,
        MIN(amount) AS minjumlah,

        MAX(amount) AS maxjumlah,
        MIN(createdat) AS transaksipertama,
        MAX(createdat) AS transaksiterakhir
    FROM transactions
    GROUP BY customerid

),
fiturkategori AS (
    -- Langkah 2: Distribusi pengeluaran antar kategori
    SELECT
        customerid,

        COUNT(DISTINCT category) AS kategoriunik,
        MODE() WITHIN GROUP (ORDER BY category) AS kategoriterbanyak,

        MAX(CASE WHEN category = 'electronics' THEN amount ELSE 0 END) AS maxelektronik
    FROM transactions
    GROUP BY customerid

),
fiturresensi AS (
    -- Langkah 3: Fitur resensi gaya RFM
    SELECT
        customerid,

        EXTRACT(EPOCH FROM (NOW() - MAX(createdat))) / 86400.0 AS harisejaktxnterakhir,

        EXTRACT(EPOCH FROM (MAX(createdat) - MIN(createdat))) / 86400.0 AS masapelangganhari

    FROM transactions
    GROUP BY customerid
)
-- Langkah 4: Gabungkan semua fitur
SELECT
    sp.customerid,

    sp.totaltransaksi,
    sp.totalbelanja,

    sp.rata2jumlah,
    sp.stddevjumlah,

    sp.maxjumlah - sp.minjumlah AS rentangjumlah,
    COALESCE(sp.stddevjumlah / NULLIF(sp.rata2jumlah, 0), 0) AS cvjumlah,

    fk.kategoriunik,
    fk.kategoriterbanyak,

    fr.harisejaktxnterakhir,
    fr.masapelangganhari,
    sp.totaltransaksi / NULLIF(fr.masapelangganhari, 0) AS frekuensitxn
FROM statistikpelanggan sp

JOIN fiturkategori fk ON sp.customerid = fk.customerid
JOIN fiturresensi fr ON sp.customerid = fr.customerid;

-- Menemukan kedalaman rantai referral (fitur mirip graf)
WITH RECURSIVE rantaireferral AS (
    -- Kasus dasar: referral langsung
    SELECT
        customerid,

        referredby,
        1 AS kedalamanrantai

    FROM customers
    WHERE referredby IS NOT NULL

    UNION ALL

    -- Langkah rekursif
    SELECT
        rr.customerid,

        c.referredby,
        rr.kedalamanrantai + 1

    FROM rantaireferral rr
    JOIN customers c ON rr.referredby = c.customerid
    WHERE c.referredby IS NOT NULL

    AND rr.kedalamanrantai < 10  -- Batas keamanan
)
SELECT customerid, MAX(kedalamanrantai) AS kedalamanreferral

FROM rantaireferral
GROUP BY customerid;

-- Buat materialized view untuk fitur pelanggan
CREATE MATERIALIZED VIEW mvfiturpelanggan AS

SELECT
    t.customerid,
    COUNT() AS totaltransaksi,

    SUM(t.amount) AS totalbelanja,
    AVG(t.amount) AS rata2transaksi,

    STDDEV(t.amount) AS stddevtransaksi,
    COUNT(DISTINCT t.category) AS keragamankategori,

    COUNT(DISTINCT DATETRUNC('day', t.createdat)) AS hariaktif,
    MAX(t.createdat) AS aktivitasterakhir,
    EXTRACT(EPOCH FROM (NOW() - MAX(t.createdat))) / 86400.0 AS haritidakaktif

FROM transactions t
GROUP BY t.customerid
WITH DATA;

-- Buat indeks untuk pencarian cepat
CREATE UNIQUE INDEX idxmvfiturpelangganid
    ON mvfiturpelanggan (customerid);


-- Segarkan view (jalankan terjadwal, misal harian via cron)
REFRESH MATERIALIZED VIEW CONCURRENTLY mvfiturpelanggan;

-- Lacak kapan view terakhir disegarkan
CREATE TABLE logpenyegaranmv (

    namaview TEXT PRIMARY KEY,
    terakhirdisegarkan TIMESTAMP NOT NULL DEFAULT NOW(),

    durasims INTEGER,
    jumlahbaris BIGINT

);

-- Fungsi penyegaran dengan pencatatan log
CREATE OR REPLACE FUNCTION segarkanviewfitur(namaview TEXT)
RETURNS VOID AS $
DECLARE
    waktumulai TIMESTAMP;

    waktuselesai TIMESTAMP;
    jmlbaris BIGINT;

BEGIN
    waktumulai := clocktimestamp();

    EXECUTE format('REFRESH MATERIALIZED VIEW CONCURRENTLY %I', namaview);
    waktuselesai := clocktimestamp();

    EXECUTE format('SELECT COUNT() FROM %I', namaview) INTO jmlbaris;

    INSERT INTO logpenyegaranmv (namaview, terakhirdisegarkan, durasims, jumlahbaris)
    VALUES (
        namaview,

        NOW(),
        EXTRACT(MILLISECONDS FROM (waktuselesai - waktumulai))::INTEGER,

        jmlbaris
    )
    ON CONFLICT (namaview)

    DO UPDATE SET
        terakhirdisegarkan = EXCLUDED.terakhirdisegarkan,

        durasims = EXCLUDED.durasims,

        jumlahbaris = EXCLUDED.jumlahbaris;

END;
$ LANGUAGE plpgsql;

-- Penggunaan
SELECT segarkanviewfitur('mvfiturpelanggan');

-- Tabel pelacakan eksperimen
CREATE TABLE eksperimenml (
    id SERIAL PRIMARY KEY,
    namaeksperimen VARCHAR(200) NOT NULL,

    tipemodel VARCHAR(100) NOT NULL,
    hyperparameter JSONB NOT NULL DEFAULT '{}',
    metrik JSONB NOT NULL DEFAULT '{}',
    konfigurasifitur JSONB,

    tag TEXT[],
    dibuatpada TIMESTAMP NOT NULL DEFAULT NOW(),
    durasipelatihandetik FLOAT
);

-- Masukkan sebuah eksperimen
INSERT INTO eksperimenml (namaeksperimen, tipemodel, hyperparameter, metrik, tag)

VALUES (
    'prediksichurnv3',

    'xgboost',
    '{"nestimators": 500, "maxdepth": 6, "learningrate": 0.01,
      "subsample": 0.8, "colsamplebytree": 0.8}',

    '{"accuracy": 0.923, "precision": 0.891, "recall": 0.876,
      "f1": 0.883, "aucroc": 0.954}',
    ARRAY['produksi', 'churn', 'v3']
);

-- Query eksperimen berdasarkan nilai hyperparameter
SELECT
    namaeksperimen,

    hyperparameter->>'learningrate' AS lr,
    metrik->>'aucroc' AS auc,

    metrik->>'f1' AS f1
FROM eksperimenml
WHERE tipemodel = 'xgboost'

  AND (hyperparameter->>'maxdepth')::int >= 5
  AND (metrik->>'aucroc')::float > 0.90

ORDER BY (metrik->>'aucroc')::float DESC;

-- Agregasi metrik antar eksperimen
SELECT
    tipemodel,

    AVG((metrik->>'aucroc')::float) AS rata2auc,

    MAX((metrik->>'aucroc')::float) AS aucterbaik,

    COUNT() AS jumlaheksperimen
FROM eksperimenml

GROUP BY tipemodel
ORDER BY rata2auc DESC;


-- Indeks GIN untuk query JSONB yang cepat
CREATE INDEX idxeksperimenhyperparams ON eksperimenml
    USING GIN (hyperparameter);
CREATE INDEX idxeksperimenmetrik ON eksperimenml

    USING GIN (metrik);

-- Buat tabel terpartisi untuk data pelatihan ML
CREATE TABLE datapelatihanml (

    id BIGSERIAL,
    vektorfitur FLOAT8[] NOT NULL,
    label INTEGER NOT NULL,
    splitdata VARCHAR(10) NOT NULL CHECK (splitdata IN ('train', 'val', 'test')),
    tanggaldibuat DATE NOT NULL DEFAULT CURRENTDATE,
    sumber VARCHAR(50),
    PRIMARY KEY (id, tanggaldibuat)

) PARTITION BY RANGE (tanggaldibuat);

-- Buat partisi bulanan
CREATE TABLE datapelatihanml202601
    PARTITION OF datapelatihanml
    FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');

CREATE TABLE datapelatihanml202602
    PARTITION OF datapelatihanml
    FOR VALUES FROM ('2026-02-01') TO ('2026-03-01');

CREATE TABLE datapelatihanml202603
    PARTITION OF datapelatihanml
    FOR VALUES FROM ('2026-03-01') TO ('2026-04-01');

-- Otomatisasi pembuatan partisi
CREATE OR REPLACE FUNCTION buatpartisibulanan(
    namatabel TEXT,

    tahun INT,
    bulan INT
) RETURNS VOID AS $
DECLARE
    namapartisi TEXT;
    tanggalmulai DATE;

    tanggalakhir DATE;
BEGIN
    namapartisi := format('%s%s%s', namatabel, tahun,
                           LPAD(bulan::TEXT, 2, '0'));
    tanggalmulai := makedate(tahun, bulan, 1);
    tanggalakhir := tanggalmulai + INTERVAL '1 month';

    EXECUTE format(
        'CREATE TABLE IF NOT EXISTS %I PARTITION OF %I
         FOR VALUES FROM (%L) TO (%L)',
        namapartisi, namatabel, tanggalmulai, tanggalakhir
    );

    RAISE NOTICE 'Partisi dibuat: %', namapartisi;

END;
$ LANGUAGE plpgsql;

-- Buat partisi untuk 12 bulan ke depan
DO $
BEGIN
    FOR m IN 1..12 LOOP
        PERFORM buatpartisibulanan('datapelatihanml', 2026, m);

    END LOOP;
END $;

-- PostgreSQL secara otomatis memangkas partisi yang tidak relevan
EXPLAIN ANALYZE
SELECT COUNT(), AVG(label::float)
FROM datapelatihanml

WHERE tanggaldibuat BETWEEN '2026-01-01' AND '2026-01-31'
  AND splitdata = 'train';

-- Hanya memindai datapelatihanml202601

Monitoring dengan pgstat

Memahami performa query sangat penting ketika pipeline ML Anda bergantung pada query database. PostgreSQL menyediakan statistik yang komprehensif melalui view pgstat.

-- Aktifkan pgstatstatements (tambahkan di postgresql.conf)
-- sharedpreloadlibraries = 'pgstatstatements'


-- Query paling lambat yang menyentuh tabel ML
SELECT
    calls,
    meanexectime::numeric(10,2) AS rata2ms,
    totalexectime::numeric(10,2) AS totalms,

    rows,
    LEFT(query, 120) AS cuplikanquery
FROM pgstatstatements
WHERE query ILIKE '%transaction%' OR query ILIKE '%ml%'

ORDER BY meanexectime DESC

LIMIT 20;

-- Statistik level tabel
SELECT
    schemaname,
    relname AS namatabel,
    seqscan,

    seqtupread,

    idxscan,
    idxtupfetch,
    ntupins AS sisipan,
    ntupupd AS pembaruan,
    ntupdel AS penghapusan,
    nlivetup AS barishidup,

    ndeadtup AS barismati,
    lastvacuum,

    lastautovacuum,
    lastanalyze

FROM pgstatusertables
WHERE relname LIKE 'ml%' OR relname LIKE 'mv%'
ORDER BY seqtupread DESC;

-- Analisis penggunaan indeks
SELECT
    indexrelname AS namaindeks,

    relname AS namatabel,
    idxscan AS kalidigunakan,
    pgsizepretty(pgrelationsize(indexrelid)) AS ukuranindeks

FROM pgstatuserindexes
WHERE schemaname = 'public'
ORDER BY idxscan DESC;


-- Rasio cache hit (seharusnya > 99%)
SELECT
    sum(heapblksread) AS heapdibaca,
    sum(heapblkshit)  AS heaphit,

    ROUND(sum(heapblkshit)  100.0 /

          NULLIF(sum(heapblkshit) + sum(heapblksread), 0), 2

    ) AS rasiocachehit

FROM pgstatiousertables;

import psycopg2
from psycopg2.extras import RealDictCursor, executevalues

import numpy as np
import pandas as pd

class KlienDatabaseML:
    """Klien database yang dioptimalkan untuk alur kerja ML."""

    def init(self, host, port, dbname, user, password):
        self.connparams = {
            "host": host,
            "port": port,
            "dbname": dbname,
            "user": user,
            "password": password,
        }
        self.conn = None

    def connect(self):
        self.conn = psycopg2.connect(self.connparams)

        self.conn.autocommit = False
        return self

    def close(self):
        if self.conn:
            self.conn.close()

    def enter(self):
        return self.connect()

    def exit(self, exctype, excval, exctb):
        if exctype:

            self.conn.rollback()
        else:
            self.conn.commit()
        self.close()

    def ambilfitur(self, daftarcustomerid: list) -> pd.DataFrame:
        """Ambil fitur yang sudah dihitung untuk daftar pelanggan."""
        query = """
            SELECT  FROM mvfiturpelanggan
            WHERE customerid = ANY(%s)

        """
        with self.conn.cursor(cursorfactory=RealDictCursor) as cur:
            cur.execute(query, (daftarcustomerid,))
            baris = cur.fetchall()
        return pd.DataFrame(baris)

    def ambildatapelatihan(self, split: str, tanggaldari: str,

                              tanggalsampai: str,
                              batas: int = None) -> pd.DataFrame:
        """Ambil data pelatihan dengan pemangkasan partisi."""
        query = """
            SELECT vektorfitur, label

            FROM datapelatihanml

            WHERE splitdata = %s
              AND tanggaldibuat BETWEEN %s AND %s

        """
        params = [split, tanggaldari, tanggalsampai]

        if batas:
            query += " LIMIT %s"
            params.append(batas)

        with self.conn.cursor() as cur:
            cur.execute(query, params)
            baris = cur.fetchall()

        fitur = np.array([b[0] for b in baris])
        label = np.array([b[1] for b in baris])
        return fitur, label

    def catateksperimen(self, nama: str, tipemodel: str,

                         hyperparams: dict, metrik: dict, tag: list):
        """Catat eksperimen ML ke database."""
        import json
        query = """
            INSERT INTO eksperimenml
                (namaeksperimen, tipemodel, hyperparameter, metrik, tag)
            VALUES (%s, %s, %s, %s, %s)
            RETURNING id
        """
        with self.conn.cursor() as cur:
            cur.execute(query, (
                nama, tipemodel,

                json.dumps(hyperparams),
                json.dumps(metrik),
                tag
            ))
            ideksperimen = cur.fetchone()[0]
        self.conn.commit()
        return ideksperimen


    def sisipkanmassaldatapelatihan(self, fitur: np.ndarray,
                                        label: np.ndarray, split: str):
        """Sisipkan data pelatihan secara massal dengan efisien."""
        data = [
            (fitur[i].tolist(), int(label[i]), split)
            for i in range(len(label))
        ]
        query = """
            INSERT INTO datapelatihanml (vektorfitur, label, splitdata)
            VALUES %s
        """
        with self.conn.cursor() as cur:
            executevalues(cur, query, data, pagesize=1000)
        self.conn.commit()


Contoh penggunaan
if name == "main":
    with KlienDatabaseML(
        host="localhost", port=5432,
        dbname="mldb", user="mluser", password="rahasia"
    ) as db:
        # Ambil fitur
        df = db.ambilfitur([1, 2, 3, 4, 5])

        print(f"Berhasil mengambil {len(df)} baris fitur pelanggan")

        # Catat eksperimen
        idexp = db.catateksperimen(

            nama="churnv4",
            tipemodel="lightgbm",

            hyperparams={"nestimators": 1000, "learningrate": 0.05},

            metrik={"aucroc": 0.961, "f1": 0.892},
            tag=["produksi", "churn"]
        )
        print(f"Eksperimen tercatat dengan ID: {idexp}")

from sqlalchemy import (
    createengine, Column, Integer, String, Float,
    DateTime, ARRAY, JSON, func, text
)
from sqlalchemy.orm import declarativebase, Session

from sqlalchemy.dialects.postgresql import JSONB
from datetime import datetime

Base = declarativebase()

class EksperimenML(Base):
    tablename = "eksperimenml"


    id = Column(Integer, primarykey=True, autoincrement=True)
    namaeksperimen = Column(String(200), nullable=False)

    tipemodel = Column(String(100), nullable=False)
    hyperparameter = Column(JSONB, default={})
    metrik = Column(JSONB, default={})
    tag = Column(ARRAY(String))
    dibuatpada = Column(DateTime, default=datetime.utcnow)

    durasipelatihandetik = Column(Float)


    def repr(self):
        return f"eksperimen} ({self.tipemodel})>"



engine = createengine(
    "postgresql://mluser:rahasia@localhost:5432/mldb",
    poolsize=10,

    maxoverflow=20,
    echo=False
)

Query eksperimen terbaik menggunakan SQLAlchemy
with Session(engine) as session:
    eksperimenterbaik = (

        session.query(EksperimenML)
        .filter(EksperimenML.tipemodel == "xgboost")
        .filter(
            EksperimenML.metrik["aucroc"].astext.cast(Float) > 0.90

        )
        .orderby(
            EksperimenML.metrik["aucroc"].astext.cast(Float).desc()

        )
        .limit(10)
        .all()
    )

    for exp in eksperimenterbaik:
        print(f"{exp.namaeksperimen}: AUC={exp.metrik['aucroc']}")

    # Gunakan SQL mentah untuk query fitur yang kompleks
    dffitur = pd.readsql(
        text("SELECT * FROM mvfiturpelanggan WHERE haritidakaktif < 30"),
        engine
    )