Data Locality

Tugas hitung frekuensi kata atas 900 GB log transaksi baru saja kita terima. Bila satu server menarik seluruh data itu dari klaster penyimpanan lewat jaringan 10 Gbps, pemindahan saja memakan waktu lebih dari dua belas menit — sebelum satu kata pun dihitung. Data locality adalah prinsip menempatkan komputasi sedekat mungkin dengan data yang diolah, sehingga yang menyeberangi jaringan adalah kode berukuran kilobyte, bukan data berukuran gigabyte. Prinsip ini menjadi asumsi dasar Hadoop dan Spark sejak awal, dan alasan utama klaster big data tidak bekerja seperti server tunggal.

Mengapa Memindahkan Data Itu Mahal #

Mesin dapat membaca dari disk lokal jauh lebih cepat daripada menarik data dari mesin lain lewat jaringan. Skrip berikut menghitung waktu memindahkan 1 TB melalui berbagai media — angka kecepatan memakai nilai tipikal perangkat keras dan jaringan data center.

python
"""Hitung waktu pindah 1 TB data lewat berbagai media."""
MEDIA = [
    ("SSD NVMe lokal", 3.0),       # GB per detik, tipikal
    ("Jaringan 10 Gbps", 1.25),    # teoretis, belum termasuk overhead protokol
    ("HDD lokal", 0.15),           # 150 MB per detik, sekuensial
    ("Jaringan 1 Gbps", 0.125),    # teoretis
]

UKURAN_GB = 1000  # 1 TB desimal

for nama, kecepatan in MEDIA:
    detik = UKURAN_GB / kecepatan
    print(f"{nama:18s} {kecepatan:5.2f} GB/s "
          f"-> {detik:6.0f} detik ({detik / 60:5.1f} menit)")

Output yang Dihasilkan:

text
SSD NVMe lokal      3.00 GB/s ->    333 detik (  5.6 menit)
Jaringan 10 Gbps    1.25 GB/s ->    800 detik ( 13.3 menit)
HDD lokal           0.15 GB/s ->   6667 detik (111.1 menit)
Jaringan 1 Gbps     0.12 GB/s ->   8000 detik (133.3 menit)

Dua pelajaran dari tabel tersebut. Pertama, jaringan 10 Gbps yang terdengar cepat tetap kalah hampir tiga kali dari SSD lokal. Kedua, jaringan 1 Gbps bahkan lebih lambat daripada HDD sekuensial — pada volume big data, waktu transfer data mendominasi runtime pipeline, bukan waktu komputasinya.

Prinsip Move Compute to Data #

Solusinya membalik arah: alih-alih menarik data ke satu titik komputasi, kirim tugas kecil ke node yang menyimpan data, jalankan pekerjaan di sana, lalu kumpulkan hasilnya. Hadoop merumuskan prinsip ini sejak desain awalnya, dan Spark melanjutkannya: scheduler menandai preferensi lokalitas (PROCESS_LOCAL, NODE_LOCAL, RACK_LOCAL) dan menunda task yang datanya berada jauh. Panduan resmi Apache Spark merinci mekanisme tersebut di dokumentasi Spark .

Yang berpindah	Ukuran tipikal	Kapan dipilih
Data mentah	ratusan GB sampai TB	hanya bila data dipakai berulang di satu tempat, misalnya load ke warehouse
Kode tugas	beberapa KB	hampir selalu: tugas dikirim ke node pemilik partisi, hasil agregat dikirim balik

Simulasi: Mengirim Data vs Mengirim Tugas #

Kita buktikan prinsip tersebut dengan simulasi nyata: tiga node (direktori terpisah) masing-masing menyimpan berkas log ~190 KB, dan tugasnya menghitung frekuensi kata. Strategi A menyalin seluruh berkas ke pusat komputasi; strategi B mengirim tugas 398 byte ke tiap node dan menerima kembali hasil ringkas. Pemakaian byte dihitung dari ukuran berkas yang benar-benar tersalin.

python
"""Bandingkan dua strategi: pindahkan data ke komputasi vs pindahkan komputasi ke data."""
import shutil
import subprocess
import sys
import random
from collections import Counter
from pathlib import Path

AKAR = Path("simulasi_lokalitas")
NODE = ["node-1", "node-2", "node-3"]
PUSAT = AKAR / "pusat-komputasi"
TUGAS = "tugas_hitung.py"

# Kode yang "dikirim" bila kita memindahkan komputasi, bukan data.
KODE_TUGAS = '''"""Tugas hitung kata: dikirim ke node tempat data berada."""
import sys
from collections import Counter
from pathlib import Path

berkas, keluaran = sys.argv[1], sys.argv[2]
hitung = Counter()
for baris in Path(berkas).read_text().splitlines():
    hitung.update(baris.split())
with Path(keluaran).open("w") as f:
    for kata, jumlah in hitung.most_common():
        f.write(f"{kata} {jumlah}\\n")
'''

KOSAKATA = ["data", "pipeline", "klaster", "partition", "worker", "stream",
            "batch", "topic", "schema", "warehouse", "lakehouse", "offset"]


def angka(n):
    """Format angka gaya Indonesia: titik sebagai pemisah ribuan."""
    return f"{n:,}".replace(",", ".")


def bangun_data():
    """Tiga node, masing-masing menyimpan satu berkas log kata (~190 KB)."""
    if AKAR.exists():
        shutil.rmtree(AKAR)
    PUSAT.mkdir(parents=True)
    acak = random.Random(42)
    for nama in NODE:
        (AKAR / nama).mkdir()
        baris = [" ".join(acak.choice(KOSAKATA) for _ in range(acak.randint(8, 14)))
                 for _ in range(2000)]
        (AKAR / nama / "log.txt").write_text("\n".join(baris) + "\n")
    (AKAR / TUGAS).write_text(KODE_TUGAS)
    (PUSAT / TUGAS).write_text(KODE_TUGAS)  # kode tugas milik pusat sendiri


def jalankan_tugas(cwd, berkas):
    """Jalankan tugas_hitung.py di cwd; kembalikan Counter hasilnya."""
    subprocess.run([sys.executable, TUGAS, berkas, "hasil.txt"], cwd=cwd, check=True)
    hitung = Counter()
    for baris in (Path(cwd) / "hasil.txt").read_text().splitlines():
        kata, jumlah = baris.rsplit(" ", 1)
        hitung[kata] = int(jumlah)
    return hitung


def strategi_pindah_data():
    """Salin seluruh berkas data ke pusat, lalu hitung kata di pusat."""
    byte_pindah = 0
    gabungan = Counter()
    for nama in NODE:
        tujuan = PUSAT / f"log_{nama}.txt"
        shutil.copy(AKAR / nama / "log.txt", tujuan)
        byte_pindah += tujuan.stat().st_size
        gabungan += jalankan_tugas(PUSAT, tujuan.name)
    print(f"  data disalin ke pusat   : {angka(byte_pindah)} byte")
    return byte_pindah, gabungan


def strategi_pindah_komputasi():
    """Kirim tugas kecil ke tiap node; node mengembalikan hanya hasil ringkas."""
    ukuran_tugas = (AKAR / TUGAS).stat().st_size
    byte_pindah = 0
    gabungan = Counter()
    for nama in NODE:
        node = AKAR / nama
        shutil.copy(AKAR / TUGAS, node / TUGAS)     # kirim kode tugas
        byte_pindah += ukuran_tugas
        hasil = jalankan_tugas(node, "log.txt")     # komputasi di node
        byte_pindah += (node / "hasil.txt").stat().st_size  # hasil balik
        gabungan += hasil
    print(f"  tugas dikirim           : 3 x {angka(ukuran_tugas)} byte")
    print(f"  total byte terkirim     : {angka(byte_pindah)} byte")
    return byte_pindah, gabungan


def ringkas(gabungan):
    atas = ", ".join(f"{k} ({v})" for k, v in gabungan.most_common(3))
    return f"total kata {angka(sum(gabungan.values()))}, teratas: {atas}"


bangun_data()
print("=== Strategi A: pindahkan DATA ke pusat komputasi ===")
byte_a, hasil_a = strategi_pindah_data()
print(f"  hasil di pusat          : {ringkas(hasil_a)}")

print("=== Strategi B: pindahkan KOMPUTASI ke node data ===")
byte_b, hasil_b = strategi_pindah_komputasi()
print(f"  hasil digabung di pusat : {ringkas(hasil_b)}")

print("=== Perbandingan ===")
print(f"  A pindah data      : {angka(byte_a)} byte")
print(f"  B pindah komputasi : {angka(byte_b)} byte ({byte_a // byte_b}x lebih sedikit)")
print(f"  hasil identik      : {hasil_a == hasil_b}")

Output yang Dihasilkan:

text
=== Strategi A: pindahkan DATA ke pusat komputasi ===
  data disalin ke pusat   : 503.640 byte
  hasil di pusat          : total kata 65.691, teratas: batch (5645), lakehouse (5641), topic (5537)
=== Strategi B: pindahkan KOMPUTASI ke node data ===
  tugas dikirim           : 3 x 398 byte
  total byte terkirim     : 1.650 byte
  hasil digabung di pusat : total kata 65.691, teratas: batch (5645), lakehouse (5641), topic (5537)
=== Perbandingan ===
  A pindah data      : 503.640 byte
  B pindah komputasi : 1.650 byte (305x lebih sedikit)
  hasil identik      : True

Mari kita telaah baris-baris kode di atas:

Baris 15-27 (KODE_TUGAS): kode tugas yang “dikirim” bila kita memindahkan komputasi — hanya 398 byte, berisi pembacaan satu berkas, penghitungan frekuensi kata, dan penulisan hasil per kata.
Baris 38-50 (bangun_data): menyiapkan tiga node berupa direktori berisi log.txt ~190 KB; generator memakai seed tetap agar ukuran berkas dan hasil hitung stabil antar eksekusi.
Baris 53-60 (jalankan_tugas): menjalankan tugas_hitung.py sebagai proses terpisah dengan cwd diarahkan ke node — meniru eksekusi di mesin tempat data berada — lalu membaca hasilnya menjadi Counter.
Baris 63-90 (dua strategi): strategi A menyalin seluruh log ke pusat lalu menghitung di pusat; strategi B mengirim kode tugas 398 byte, menjalankan komputasi di node, lalu menerima hasil ringkas — byte yang berpindah dihitung dari ukuran berkas yang benar-benar tersalin.
Baris 107-110: ringkasan akhir — hasil identik : True membuktikan kedua strategi menghasilkan jawaban sama persis dengan biaya jaringan berbeda; pada skala produksi rasionya jauh lebih ekstrem karena tugas tetap beberapa KB sementara partisi data diukur dalam gigabyte.

Menjaga Lokalitas Saat Mendesain Pipeline #

Lokalitas bekerja karena data sudah terbagi menjadi partition — unit penempatan yang dibahas pada Chapter Partitioning dan Sharding. Scheduler cukup mencocokkan setiap task dengan node pemilik partisinya, tanpa perlu memindahkan byte dalam jumlah besar. Dua situasi mengikis lokalitas ini. Pertama, data skew: satu partisi jauh lebih besar dari yang lain, sehingga satu worker sibuk lama sementara sisanya menganggur. Kedua, shuffle: operasi seperti join dua dataset besar memaksa data berpindah antar node karena key yang cocok tersebar di mana-mana.

Dengan data diam di tempat dan tugas yang mendekat, satu pertanyaan tersisa: bagaimana membagi pekerjaan itu ke banyak worker agar selesai serentak. Chapter Parallel Processing membahas pembagian partisi kerja, benchmark paralel nyata, dan Hukum Amdahl yang membatasi percepatan maksimalnya.