Data harian sebuah perusahaan ritel bisa mencapai jutaan baris per hari, dan file histori akhirnya berukuran gigabyte. Memuat file semacam itu seluruhnya ke RAM bukan pilihan. Strateginya ada tiga: memotong data menjadi bagian (chunking), mengalirkan transformasi tanpa parkir data di memori (streaming), dan menggabungkan hasil antara secara bertahap (pola map-reduce). Kita praktikkan ketiganya di Python murni, lalu menutup dengan panduan kapan saatnya berpindah ke pandas, DuckDB, atau Spark.

Tiga Pola Dasar #

Sebelum masuk ke kode, kita petakan dulu peran masing-masing pola:

  • Streaming: data dibaca satu baris pada satu waktu dari sumber, diproses, lalu dilupakan — memori hanya menampung satu baris dan hasil agregat. Ini pola yield dari Chapter Iterators dan Generators.
  • Chunking: aliran dipotong menjadi blok berukuran tetap, misalnya 25.000 baris. Blok diproses satu per satu; memori puncak mengikuti ukuran blok, bukan ukuran file.
  • Map-reduce sederhana: tahap map menormalkan dan menyaring tiap baris menjadi pasangan kunci-nilai; tahap reduce menggabungkan nilai per kunci. Dengan chunking, reduce berjalan dua lapis — parsial per blok, lalu penggabungan akhir antar blok.

Ketiganya digabung dalam satu studi kasus berikut.

Studi Kasus: Agregasi Bertahap dari Iterator Record #

Kita mengolah penjualan.csv berisi 120.000 baris dengan beberapa baris kotor: kolom kota kosong dan total negatif. Targetnya satu laporan: jumlah transaksi dan total pendapatan per kota. Aliran data dipotong lima blok, tiap blok diagregasi parsial, lalu hasil parsial digabung.

python 
import csv
import random
import time
from collections import defaultdict
from itertools import islice
import tracemalloc

N = 120_000
CHUNK = 25_000

# 1. Siapkan sumber data: 120.000 baris dengan beberapa baris kotor
rng = random.Random(7)
kota_pool = ["jakarta", "bandung", "surabaya", "medan", "denpasar", ""]
with open("penjualan.csv", "w", newline="") as f:
    tulis = csv.writer(f)
    tulis.writerow(["id", "kota", "total"])
    for i in range(1, N + 1):
        tulis.writerow([i, rng.choice(kota_pool), rng.randrange(-20_000, 250_000, 500)])


# 2. Streaming: iterator record dari file (satu baris di memori pada satu waktu)
def stream_record(nama_file):
    with open(nama_file, newline="") as f:
        for baris in csv.DictReader(f):
            yield baris


# 3. Chunking: potong aliran menjadi blok
def potong_aliran(aliran, ukuran):
    while True:
        blok = list(islice(aliran, ukuran))
        if not blok:
            return
        yield blok


# 4. Map: normalisasi satu record
def map_bersih(record):
    kota = record["kota"].strip().upper()
    total = int(record["total"])
    if kota == "" or total <= 0:
        return None
    return (kota, total)


# 5. Reduce: agregasi parsial per blok
def reduce_blok(blok):
    parsial = defaultdict(lambda: [0, 0])
    for record in blok:
        hasil_map = map_bersih(record)
        if hasil_map is None:
            continue
        kota, total = hasil_map
        parsial[kota][0] += 1
        parsial[kota][1] += total
    return dict(parsial)


# 6. Jalankan pipeline bertahap dan gabungkan hasil parsial
tracemalloc.start()
mulai = time.perf_counter()

parsial_list = []
for nomor, blok in enumerate(potong_aliran(stream_record("penjualan.csv"), CHUNK), 1):
    parsial_list.append(reduce_blok(blok))
    print(f"blok {nomor}: {len(blok)} baris dibaca")

final = defaultdict(lambda: [0, 0])
for parsial in parsial_list:
    for kota, (n, s) in parsial.items():
        final[kota][0] += n
        final[kota][1] += s

durasi = time.perf_counter() - mulai
puncak = tracemalloc.get_traced_memory()[1]
tracemalloc.stop()

final = dict(sorted(final.items()))
print(f"-- hasil akhir ({durasi:.2f} detik) --")
for kota, (n, s) in final.items():
    print(f"{kota:10s} {n:6d} transaksi total Rp{s:,}")
print(f"puncak memori: {puncak / 1024:.1f} KB")

Output yang Dihasilkan:

text 
blok 1: 25000 baris dibaca
blok 2: 25000 baris dibaca
blok 3: 25000 baris dibaca
blok 4: 25000 baris dibaca
blok 5: 20000 baris dibaca
-- hasil akhir (1.00 detik) --
BANDUNG     18629 transaksi total Rp2,336,736,000
DENPASAR    18366 transaksi total Rp2,299,657,000
JAKARTA     18566 transaksi total Rp2,312,285,000
MEDAN       18286 transaksi total Rp2,295,067,500
SURABAYA    18575 transaksi total Rp2,328,485,000
puncak memori: 15922.6 KB

Mari kita telaah baris-baris kode di atas:

  1. Baris 12-18 (pembuatan data): seed tetap random.Random(7) membuat hasil uji dapat diulang. Kolom kota_pool memuat string kosong dan randrange memungkinkan total negatif — dua jenis baris kotor yang harus tersaring.
  2. Baris 22-25 (stream_record): generator pembaca file. Iterator ini adalah sumber aliran; hanya satu baris yang berupa dictionary pada satu waktu.
  3. Baris 29-34 (potong_aliran): islice(aliran, ukuran) mengambil sampai ukuran item dari aliran lalu berhenti. list(...) memadatkannya menjadi satu blok; bila blok kosong, generator selesai. Hasilnya: lima blok (25.000 × 4, lalu 20.000 sisa).
  4. Baris 38-43 (map_bersih): tahap map menormalkan kota ke huruf besar dan membuang baris kotor dengan None. Baris valid menjadi pasangan (kota, total).
  5. Baris 47-56 (reduce_blok): tahap reduce lapis pertama — agregasi parsial di dalam satu blok. defaultdict(lambda: [0, 0]) menyimpan pasangan [jumlah, total] per kota.
  6. Baris 63-72 (penggabungan akhir): reduce lapis kedua menggabungkan lima hasil parsial menjadi final. Struktur map-reduce-nya: map_bersih dipanggil 120.000 kali, reduce_blok lima kali, penggabungan akhir satu kali.

Kapan Berpindah dari Python Murni #

Python murni unggul pada transformasi per baris yang fleksibel. Namun begitu analitik menjadi rumit — banyak GROUP BY, join, window function — SQL jauh lebih ringkas. Panduan berikut menentukan titik pindahnya:

Kondisi dataPilihan yang tepatCatatan
File ratusan MB; transformasi baris demi barisPython stdlib (csv + generator)Pola chapter ini; tanpa dependensi tambahan
Analitik tabular interaktif; data muat di RAMpandasDataFrame kaya fitur; memori mengikuti ukuran data
Query SQL pada file besar, termasuk melebihi RAMDuckDBSatu proses; langsung membaca CSV/Parquet
Terabyte; klaster multi-node; batch terjadwalSparkTerdistribusi ke banyak mesin; biaya operasional paling tinggi

Sebagai bukti, laporan yang sama ditulis ulang sebagai satu query DuckDB terhadap file penjualan.csv yang sama.

python 
import duckdb

hasil = duckdb.sql(
    """
    SELECT upper(trim(kota)) AS kota,
           count(*) AS jumlah,
           sum(total) AS total
    FROM read_csv_auto('penjualan.csv')
    WHERE trim(kota) <> '' AND total > 0
    GROUP BY upper(trim(kota))
    ORDER BY kota
    """
).fetchall()

for kota, jumlah, total in hasil:
    print(f"{kota:10s} {jumlah:6d} transaksi total Rp{total:,}")

Output yang Dihasilkan:

text 
BANDUNG     18629 transaksi total Rp2,336,736,000
DENPASAR    18366 transaksi total Rp2,299,657,000
JAKARTA     18566 transaksi total Rp2,312,285,000
MEDAN       18286 transaksi total Rp2,295,067,500
SURABAYA    18575 transaksi total Rp2,328,485,000

Angka pada output DuckDB sama persis dengan hasil Python murni — 18.629 transaksi Bandung dengan total Rp2.336.736.000, dan seterusnya untuk lima kota. Sepuluh baris SQL menggantikan seluruh pipeline chunking di atas. Kekuatan Python murni tetap pada kontrol penuh: validasi kustom, panggilan API per baris, atau format yang belum dikenali pembaca SQL.

Rujukan resmi untuk pendalaman: itertools , dokumentasi pandas , dan dokumentasi DuckDB . Pipeline yang andal tidak cukup hanya berjalan — ia harus terbukti benar. Chapter Pengujian Kode Python berikutnya membangun jaring pengamannya dengan pytest.