Distributed Query Processing

"""Ch8: Distributed query processing — simulasi coordinator/worker di satu mesin.

1 juta pesanan TokoKita dipecah jadi 8 direktori hive-partition (kota=<nama>);
koordinator membagi shard ke 2 worker, tiap worker menghitung agregasi parsial
(DuckDB lokal), koordinator menggabungkan hasil parsial.
"""
import glob
import os
import shutil
import time
from multiprocessing import Pool

import duckdb

KOTA = ["Bandung", "Denpasar", "Jakarta", "Medan",
        "Palembang", "Semarang", "Surabaya", "Yogyakarta"]
N_PER_KOTA = 125_000
GLOB = "shards/*/*.parquet"

if os.path.exists("shards"):
    shutil.rmtree("shards")

con = duckdb.connect()
con.execute(f"""
CREATE OR REPLACE TABLE pesanan AS
SELECT
    i AS id,
    CASE (i - 1) % 8
        WHEN 0 THEN 'Bandung' WHEN 1 THEN 'Denpasar' WHEN 2 THEN 'Jakarta'
        WHEN 3 THEN 'Medan'   WHEN 4 THEN 'Palembang' WHEN 5 THEN 'Semarang'
        WHEN 6 THEN 'Surabaya' ELSE 'Yogyakarta' END AS kota,
    'pelanggan-' || (i % 10000)::VARCHAR AS pelanggan,
    15000.0 + ((i % 500) * 250.0)::DOUBLE AS nilai
FROM range(1, {len(KOTA) * N_PER_KOTA + 1}) AS t(i)
""")
for kota in KOTA:
    os.makedirs(f"shards/kota={kota}", exist_ok=True)
    con.execute(f"""
        COPY (SELECT * FROM pesanan WHERE kota = '{kota}')
        TO 'shards/kota={kota}/data.parquet' (FORMAT parquet)
    """)

berkas_shard = sorted(glob.glob(GLOB))
total_baris = con.execute("SELECT count(*) FROM pesanan").fetchone()[0]
total_byte = sum(os.path.getsize(f) for f in berkas_shard)
print(f"shard ditulis: {len(berkas_shard)} direktori, total {total_byte:,} byte, "
      f"{total_baris:,} baris")


def agregasi_parsial(berkas):
    """Worker: agregasi parsial per pelanggan pada satu shard."""
    koneksi = duckdb.connect()
    hasil = koneksi.execute(
        f"SELECT pelanggan, count(*) AS jumlah, sum(nilai) AS total "
        f"FROM read_parquet('{berkas}') GROUP BY pelanggan"
    ).fetchall()
    koneksi.close()
    return hasil


def gabung(hasil_parsial):
    """Coordinator: gabungkan hasil parsial per pelanggan lintas shard."""
    total = {}
    for batch in hasil_parsial:
        for pelanggan, jumlah, nilai in batch:
            agg = total.setdefault(pelanggan, [0, 0.0])
            agg[0] += jumlah
            agg[1] += nilai
    return total


# --- 1. sekuensial: satu worker memproses 8 shard berurutan ---
agregasi_parsial(berkas_shard[0])  # pemanasan
t0 = time.perf_counter()
parsial = [agregasi_parsial(f) for f in berkas_shard]
hasil_sekuensial = gabung(parsial)
durasi_sekuensial = time.perf_counter() - t0
print(f"sekuensial 1 worker : {durasi_sekuensial:.3f} detik")

if __name__ == "__main__":
    # --- 2. paralel: coordinator membagi 8 shard ke 2 worker ---
    with Pool(2) as pool:
        pool.map(agregasi_parsial, berkas_shard[:2])  # pemanasan
        t0 = time.perf_counter()
        parsial_paralel = pool.map(agregasi_parsial, berkas_shard)
        hasil_paralel = gabung(parsial_paralel)
        durasi_paralel = time.perf_counter() - t0
    print(f"paralel 2 worker    : {durasi_paralel:.3f} detik")
    print(f"speedup             : {durasi_sekuensial / durasi_paralel:.2f}x")
    print(f"hasil identik       : {hasil_sekuensial == hasil_paralel}")

    # --- 3. satu query DuckDB membaca seluruh glob ---
    q_glob = ("SELECT pelanggan, count(*), sum(nilai) "
              f"FROM read_parquet('{GLOB}') GROUP BY pelanggan")
    con.execute(q_glob).fetchall()  # pemanasan
    t0 = time.perf_counter()
    con.execute(q_glob).fetchall()
    print(f"DuckDB glob 1 query : {time.perf_counter() - t0:.3f} detik")

    # --- 4. partition pruning: filter kota hanya membuka 1 dari 8 shard ---
    q_prune = (f"SELECT kota, count(*), round(sum(nilai), 2) AS total "
               f"FROM read_parquet('{GLOB}', hive_partitioning=true) "
               f"WHERE kota = 'Medan' GROUP BY kota")
    con.execute(q_prune).fetchall()
    t0 = time.perf_counter()
    hasil_prune = con.execute(q_prune).fetchall()
    print(f"filter kota='Medan' : {time.perf_counter() - t0:.3f} detik -> {hasil_prune}")

    plan = con.execute("EXPLAIN ANALYZE " + q_prune).fetchall()[0][1]
    lines = [l.replace("│", "").strip() for l in plan.splitlines()]
    mulai = next(i for i, l in enumerate(lines) if "TABLE_SCAN" in l)
    print()
    print("kutipan TABLE_SCAN (filter kota='Medan'):")
    for l in lines[mulai:mulai + 20]:
        if l:
            print(l)

shard ditulis: 8 direktori, total 4,703,703 byte, 1,000,000 baris
sekuensial 1 worker : 0.108 detik
paralel 2 worker    : 0.071 detik
speedup             : 1.52x
hasil identik       : True
DuckDB glob 1 query : 0.012 detik
filter kota='Medan' : 0.002 detik -> [('Medan', 125000, 9625000000.0)]

kutipan TABLE_SCAN (filter kota='Medan'):
TABLE_SCAN
────────────────────
Function:
READ_PARQUET
Projections:
kota
nilai
File Filters:
(kota = 'Medan')
Scanning Files: 1/8
Total Files Read: 1
Filename(s):
shards/kota=Medan/data
.parquet