Analisis Data Berskala Besar

Ketika dataset berukuran terabyte tidak muat di RAM, algoritma in-memory tradisional menjadi tidak relevan. Hadoop, Spark, dan sistem distributed lainnya lahir untuk menjawab tantangan ini, tetapi fondasinya tetap konsep DSA klasik: external sort dengan chunk-merge, hash-based aggregation, dan pola MapReduce. Pada Chapter ini kita akan mengimplementasikan external sort sederhana dengan dua fase — split data menjadi chunk yang muat di memori, sort masing-masing chunk, lalu gabungkan dengan k-way merge menggunakan min-heap. Setelah sorting, hash map dipakai untuk agregasi count-per-nilai. Implementasi ini menunjukkan bahwa prinsip DSA yang sama yang bekerja untuk data kecil dapat diskalakan ke big data dengan strategi yang tepat.

Tantangan Data Berskala Besar #

Data besar membawa tiga tantangan khas yang tidak ada pada data kecil:

Tantangan	Implikasi	Strategi Mitigasi
Tidak muat di RAM	Algoritma in-memory gagal	External sort, streaming
I/O disk dominan	CPU idle menunggu disk	Sequential I/O, batching
Distribusi data	Single machine terlalu lambat	Sharding, parallel processing

flowchart LR A[Big Data] --> B[Phase 1: Chunk] B --> C1[Chunk 1 sorted] B --> C2[Chunk 2 sorted] B --> C3[Chunk k sorted] C1 --> D[Phase 2: k-way merge] C2 --> D C3 --> D D --> E[Phase 3: Hash aggregate] E --> F[Hasil analisis]

External Sort: Aturan dan Mekanisme #

Aturan #

External sort memecah data besar menjadi chunk berukuran M yang muat di memori. Setiap chunk diurutkan secara in-memory (std::sort O(M log M)). Kemudian, semua chunk digabungkan dengan k-way merge menggunakan min-heap. Output adalah data terurut penuh, meskipun data asli terlalu besar untuk RAM.

Mekanisme di Balik Layar #

Phase 1 — Chunk + Sort:

Baca M elemen dari input ke vector.
Sort vector dengan std::sort (O(M log M)).
Tulis chunk terurut ke disk (atau simpan di memori untuk demo).
Ulangi hingga data habis. Total: n/M chunk.

Phase 2 — K-way Merge:

Inisialisasi min-heap dengan elemen pertama setiap chunk.
Pop elemen terkecil, tambahkan ke output.
Push elemen berikutnya dari chunk yang sama ke heap.
Ulangi hingga semua elemen terproses.

Phase 3 — Hash Aggregation:

Iterasi data terurut (atau stream dari hasil merge).
Untuk setiap elemen, increment hash_map[elem].
Hasil: map nilai → frekuensi kemunculan.

Kompleksitas #

Operasi	Time	Space	I/O Disk
Phase 1 (sort chunks)	O(n log M) total	O(M) per chunk	O(n) write sequential
Phase 2 (k-way merge)	O(n log k), k = n/M	O(k) heap	O(n) read + write
Phase 3 (aggregate)	O(n) rata-rata	O(unique values)	O(n) read
Total	O(n log M + n log k)	O(M + k)	O(n) per pass

Untuk skala besar, I/O disk adalah bottleneck — total waktu didominasi oleh jumlah pass disk. Multi-level merge mengurangi pass dari O(k) ke O(log k / log M).

Kode: External Sort Mini #

cpp
#include <iostream>
#include <vector>
#include <algorithm>
#include <queue>
#include <unordered_map>
#include <string>
#include <iomanip>
#include <random>
using namespace std;

const int CHUNK_SIZE = 5;   // demo kecil agar trace jelas
const int DATA_SIZE = 20;   // total data 20 angka

// Generate data acak ke vector
vector<int> generateData(int n, unsigned seed) {
    mt19937 rng(seed);
    uniform_int_distribution<int> dist(1, 100);
    vector<int> data(n);
    for (int i = 0; i < n; ++i) data[i] = dist(rng);
    return data;
}

// Phase 1: pecah data jadi chunk, sort masing-masing chunk
vector<vector<int>> makeChunks(const vector<int>& data, int chunk_size) {
    vector<vector<int>> chunks;
    for (size_t i = 0; i < data.size(); i += chunk_size) {
        vector<int> chunk(data.begin() + i,
                          data.begin() + min(i + (size_t)chunk_size, data.size()));
        sort(chunk.begin(), chunk.end());
        chunks.push_back(move(chunk));
    }
    return chunks;
}

// Phase 2: k-way merge menggunakan min-heap
vector<int> kwayMerge(const vector<vector<int>>& chunks) {
    struct Item {
        int val;
        int chunk_idx;
        int elem_idx;
        bool operator>(const Item& o) const { return val > o.val; }
    };
    priority_queue<Item, vector<Item>, greater<Item>> pq;

    for (int i = 0; i < (int)chunks.size(); ++i) {
        if (!chunks[i].empty()) {
            pq.push({chunks[i][0], i, 0});
        }
    }

    vector<int> result;
    while (!pq.empty()) {
        Item it = pq.top(); pq.pop();
        result.push_back(it.val);
        int next = it.elem_idx + 1;
        if (next < (int)chunks[it.chunk_idx].size()) {
            pq.push({chunks[it.chunk_idx][next], it.chunk_idx, next});
        }
    }
    return result;
}

// Phase 3: hash-based aggregation (mis. hitung frekuensi)
unordered_map<int,int> hashAggregate(const vector<int>& sorted) {
    unordered_map<int,int> counts;
    for (int v : sorted) counts[v]++;
    return counts;
}

int main() {
    cout << "=== Analisis Data Berskala Besar ===\n\n";
    cout << "Demo: external sort + hash aggregation\n";
    cout << "Data size = " << DATA_SIZE << ", chunk size = " << CHUNK_SIZE << "\n\n";

    auto data = generateData(DATA_SIZE, 123);

    cout << "Data asli (mentah):\n  ";
    for (size_t i = 0; i < data.size(); ++i) {
        cout << data[i];
        if (i+1 < data.size()) cout << " ";
    }
    cout << "\n\n";

    cout << "Phase 1: Split jadi chunk, sort tiap chunk:\n";
    auto chunks = makeChunks(data, CHUNK_SIZE);
    for (size_t i = 0; i < chunks.size(); ++i) {
        cout << "  Chunk " << i << ": ";
        for (size_t j = 0; j < chunks[i].size(); ++j) {
            cout << chunks[i][j];
            if (j+1 < chunks[i].size()) cout << " ";
        }
        cout << "\n";
    }
    cout << "\n";

    cout << "Phase 2: K-way merge menggunakan min-heap:\n";
    auto sorted = kwayMerge(chunks);
    cout << "  Hasil sorted: ";
    for (size_t i = 0; i < sorted.size(); ++i) {
        cout << sorted[i];
        if (i+1 < sorted.size()) cout << " ";
    }
    cout << "\n\n";

    cout << "Phase 3: Hash-based aggregation (count per nilai):\n";
    auto counts = hashAggregate(sorted);
    vector<pair<int,int>> sorted_counts(counts.begin(), counts.end());
    sort(sorted_counts.begin(), sorted_counts.end());

    cout << "  " << left << setw(10) << "Nilai" << "Frekuensi" << "\n";
    cout << "  ----------|----------\n";
    for (const auto& kv : sorted_counts) {
        cout << "  " << left << setw(10) << kv.first << kv.second << "\n";
    }

    cout << "\n=== Ringkasan Kompleksitas ===\n";
    cout << "External sort (n data, memori M):\n";
    cout << "  I/O passes: O(n/M) chunks, masing-masing sort O(M log M)\n";
    cout << "  K-way merge: O(n log k), k = n/M chunks\n";
    cout << "  Total I/O: O((n/M) * (n/M)) = O(n^2/M^2) passes (naive)\n";
    cout << "  Dengan multi-level merge: O(n log(n/M) / log M)\n";
    return 0;
}

Output yang Dihasilkan:

text
=== Analisis Data Berskala Besar ===

Demo: external sort + hash aggregation
Data size = 20, chunk size = 5

Data asli (mentah):
  70 72 29 43 23 70 56 72 72 50 43 79 99 42 69 58 49 14 40 41

Phase 1: Split jadi chunk, sort tiap chunk:
  Chunk 0: 23 29 43 70 72
  Chunk 1: 50 56 70 72 72
  Chunk 2: 42 43 69 79 99
  Chunk 3: 14 40 41 49 58

Phase 2: K-way merge menggunakan min-heap:
  Hasil sorted: 14 23 29 40 41 42 43 43 49 50 56 58 69 70 70 72 72 72 79 99

Phase 3: Hash-based aggregation (count per nilai):
  Nilai     Frekuensi
  ----------|----------
  14        1
  23        1
  29        1
  40        1
  41        1
  42        1
  43        2
  49        1
  50        1
  56        1
  58        1
  69        1
  70        2
  72        3
  79        1
  99        1

=== Ringkasan Kompleksitas ===
External sort (n data, memori M):
  I/O passes: O(n/M) chunks, masing-masing sort O(M log M)
  K-way merge: O(n log k), k = n/M chunks
  Total I/O: O((n/M) * (n/M)) = O(n^2/M^2) passes (naive)
  Dengan multi-level merge: O(n log(n/M) / log M)

Bedah Kode Baris-per-Baris #

for (size_t i = 0; i < data.size(); i += chunk_size): iterasi data dengan langkah chunk_size. Setiap iterasi mengambil satu chunk. size_t dipakai untuk menghindari warning signedness saat membandingkan dengan data.size().
vector<int> chunk(data.begin() + i, data.begin() + min(i + chunk_size, data.size()));: konstruktor vector dari iterator range. min memastikan chunk terakhir tidak melewati akhir data — penting ketika data.size() tidak kelipatan chunk_size.
sort(chunk.begin(), chunk.end());: sort chunk in-memory. std::sort introsort (hybrid quicksort + heapsort + insertion sort) menjamin O(M log M) worst case.
chunks.push_back(move(chunk));: move menghindari copy expensive vector — ownership data dipindahkan ke vector chunks. Ini adalah optimasi C++ modern yang krusial untuk data besar.
struct Item { int val; int chunk_idx; int elem_idx; bool operator>(const Item& o) const { return val > o.val; } };: item heap menyimpan nilai, indeks chunk asal, dan indeks elemen dalam chunk. Operator > mengubah max-heap default menjadi min-heap.
priority_queue<Item, vector<Item>, greater<Item>> pq;: min-heap dari Item. greater<Item> sebagai comparator memastikan item dengan val terkecil di-top().
for (int i = 0; i < (int)chunks.size(); ++i) if (!chunks[i].empty()) pq.push({chunks[i][0], i, 0});: inisialisasi heap dengan elemen pertama setiap chunk. Cast (int) menghindari warning perbandingan signedness.
Item it = pq.top(); pq.pop(); result.push_back(it.val);: ambil elemen terkecil, tambahkan ke output. Heap merestuktur diri setelah pop.
int next = it.elem_idx + 1; if (next < (int)chunks[it.chunk_idx].size()) pq.push({...});: push elemen berikutnya dari chunk yang sama. Bila chunk habis, heap otomatis menyusut.
for (int v : sorted) counts[v]++;: hash aggregation. Operator [] pada unordered_map menginisialisasi entri baru ke 0, lalu ++ mengincrement. Lookup O(1) rata-rata.

Pola MapReduce #

MapReduce adalah generalisasi external sort + hash aggregation untuk distributed cluster. Konsepnya:

flowchart LR A[Input split] --> M1[Map 1] A --> M2[Map 2] A --> M3[Map k] M1 --> S[Shuffle by key] M2 --> S M3 --> S S --> R1[Reduce 1] S --> R2[Reduce 2] S --> R3[Reduce m] R1 --> O[Output] R2 --> O R3 --> O

Fase	Setara dengan	Output
Map	Apply function per elemen	(key, value) pairs
Shuffle	Hash-based grouping by key	(key, [v1, v2, …])
Reduce	Aggregate per key	Hasil per key

Contoh: menghitung frekuensi kata di Wikipedia.

Map: tiap mapper menghitung frekuensi lokal pada bagian dokumennya → (word, count).
Shuffle: semua pasangan dengan word yang sama dikelompokkan.
Reduce: jumlahkan count per word → (word, total_count).

Strategi untuk Skala Besar #

Strategi	Tujuan	Teknik
Partitioning	Distribusi beban	Hash partitioning, range partitioning
Sampling	Estimasi tanpa full scan	Reservoir sampling
Bloom filter	Cek membership hemat memori	Probabilistic, false positive
HyperLogLog	Estimasi cardinality	Probabilistic, sub-linear memory
Count-Min Sketch	Frekuensi approximate	Sub-linear memory
Streaming	Single pass	Sliding window, exponential decay

Aplikasi Nyata #

Sistem	Teknik Inti	Skala
Hadoop MapReduce	External sort + hash agg	Petabyte
Apache Spark	RDD, in-memory cache	Terabyte – petabyte
Google BigQuery	Columnar + Dremel	Petabyte
Elasticsearch	Inverted index + sharding	Terabyte
ClickHouse	Vectorized execution	Petabyte

Ringkasan #

External sort membagi data besar menjadi chunk, sort in-memory, lalu k-way merge.
K-way merge menggunakan min-heap — efisien O(n log k).
Hash-based aggregation memberi output count-per-key dalam O(n) rata-rata.
MapReduce adalah generalisasi external sort untuk cluster distributed.
Untuk skala besar, pertimbangkan teknik approximate (Bloom filter, HyperLogLog, Count-Min Sketch) untuk menghemat memori.
Pemilihan antara sort-based dan hash-based aggregation bergantung pada kebutuhan urutan dan ukuran data.