MapReduce — map, shuffle, reduce trên hàng nghìn máy

TL;DR: MapReduce là mô hình tính toán phân tán chia job lớn thành ba pha: Map (mỗi worker biến đổi một mảnh input thành cặp key-value), Shuffle (hệ thống gom và sắp xếp mọi giá trị cùng key về đúng một reducer — đây là pha đắt nhất và thực chất là external sort chạy phân tán), Reduce (reducer tổng hợp danh sách giá trị của mỗi key thành kết quả cuối). Năm 2004, Google chạy mô hình này trên cluster khoảng 1.800 máy để sort 1 TB dữ liệu trong 891 giây và dựng chỉ mục từ khoảng 20 TB HTML đã crawl. Shuffle chiếm khoảng 1/3 tổng thời gian job điển hình và là nút thắt chính khi dữ liệu lớn.

Năm 2004, nhóm kỹ sư Google đứng trước bài toán không giải được bằng một máy: dựng chỉ mục tìm kiếm từ khoảng 20 TB tài liệu HTML đã crawl được. Không có máy đơn nào đủ RAM, đủ đĩa, hay đủ CPU để xử lý khối đó trong thời gian hợp lý. Cần nghìn máy — nhưng nghìn máy thì phối hợp thế nào?

Câu trả lời của Dean và Ghemawat trong bài báo OSDI 2004 là lập trình viên không cần viết code phối hợp. Họ chỉ cần khai báo hai hàm: map (biến đổi từng dòng thành cặp key-value) và reduce (tổng hợp mọi giá trị của cùng một key). Framework lo phần còn lại: chia input, chạy song song, gom kết quả. Kết quả: cluster khoảng 1.800 máy sort 1 TB dữ liệu trong 891 giây (~15 phút).

Bài này giải thích ba pha map → shuffle → reduce từ góc độ cơ chế — đặc biệt là phase shuffle, thứ ít được giải thích nhưng chiếm khoảng 1/3 thời gian job và thực chất là external sort chạy phân tán mà bạn đã biết nguyên lý từ bài thuật toán.

1. Analogy — bưu cục phân loại thư toàn quốc

Hình dung một chiến dịch gửi 100 triệu lá phiếu về trung ương để đếm. Không có bưu cục nào xử lý nổi 100 triệu lá một mình, nên cả hệ thống chia thành ba tầng:

Tầng 1 — Nhân viên địa phương (Map): mỗi nhân viên nhận một túi thư nhỏ, mở từng phong bì, xem tỉnh đích, dán nhãn (tỉnh, lá phiếu). Nhân viên không đếm — chỉ phân loại và dán nhãn.

Tầng 2 — Xe tải phân phối (Shuffle): xe tải gom tất cả lá được dán cùng tỉnh, chở về một trung tâm tỉnh duy nhất, và sắp xếp thứ tự trước khi đặt lên bàn đếm. Đây là công đoạn nặng nhất.

Tầng 3 — Nhân viên tỉnh (Reduce): mỗi nhân viên tỉnh nhận đúng chồng lá phiếu của tỉnh mình (đã sắp sẵn thứ tự), đếm, ghi kết quả.

2. Tổng quan ba pha — map, shuffle, reduce

Một MapReduce job xử lý input theo thứ tự cố định: Map song song → Shuffle tự động → Reduce song song.

flowchart LR
  INPUT["Input (HDFS)"] --> M1["Mapper 1"]
  INPUT --> M2["Mapper 2"]
  INPUT --> M3["Mapper 3"]

  M1 -->|"key-value"| SH["SHUFFLE<br/>sort + route"]
  M2 -->|"key-value"| SH
  M3 -->|"key-value"| SH

  SH -->|"key A, vals"| R1["Reducer A"]
  SH -->|"key B, vals"| R2["Reducer B"]

  R1 --> OUT["Output (HDFS)"]
  R2 --> OUT

Điểm then chốt: Shuffle là tầng trung gian duy nhất nối mọi mapper với mọi reducer. Mọi output của mapper đều chạy qua Shuffle trước khi tới reducer — không có đường tắt.

3. Phase Map — biến đổi từng mảnh input thành cặp key-value

Framework chia input thành nhiều split (mảnh) — thường tương ứng một block HDFS khoảng 128 MB. Mỗi split chạy trên một Mapper song song với các Mapper khác.

Hàm map do người dùng viết, nhận từng dòng input, phát ra một hoặc nhiều cặp (key, value):

-- Job: dem so don hang theo san pham
function map(line):
    order <- parse(line)
    emit(order.product_id, 1)    -- moi dong emit mot cap (san pham, 1)

Với job "đếm số đơn theo sản phẩm" trên 40 triệu dòng log: mỗi mapper nhận một mảnh dữ liệu và phát ra các cặp ("product_123", 1), ("product_456", 1), v.v. Tất cả chạy song song — N mapper cùng lúc.

Sau khi mapper phát ra cặp key-value, framework không gửi thẳng cho reducer mà buffer trong RAM (thường 100 MB), rồi khi buffer đầy thì spill (tràn) ra đĩa thành một sorted run. Đây là bước đầu của external sort — giải thích kỹ ở section tiếp theo.

4. Phase Shuffle — external sort phân tán (điểm khoá)

Đây là pha ít được giải thích nhất nhưng là trái tim của MapReduce. Mục tiêu: đảm bảo mọi giá trị có cùng key tập hợp tại đúng một reducer, đến nơi theo thứ tự đã sắp xếp.

4.1 Phía Mapper — sort và partition cục bộ

Khi buffer đầy, framework thực hiện:

1. Sort các entry trong buffer theo (partition, key) — partition xác định reducer nào nhận.
2. Spill (ghi tràn) ra đĩa thành một sorted run.
3. Cuối cùng, merge tất cả sorted run của mapper đó thành một file duy nhất.

HashPartitioner mặc định: partition(key) = hash(key) % numReducers. Mọi record cùng key → cùng partition → cùng reducer.

flowchart TD
  KV["Key-value from mapper"] --> BUF["RAM buffer (100 MB)"]
  BUF -->|full| SORT["Sort: partition + key"]
  SORT --> DISK["Sorted run on disk"]
  DISK --> MERGE["Merge all runs into 1 file"]
  MERGE --> NET["HTTP transfer to reducer"]

4.2 Phía Reducer — fetch và merge nhiều sorted run

Reducer kéo (fetch) phần partition của mình từ tất cả mapper qua HTTP. Nó nhận nhiều sorted run (một từ mỗi mapper), rồi merge chúng thành một dòng key-value sorted toàn cục — đúng là k-way merge của external sort:

-- Tu Mapper 1: ("product_123", 1), ("product_456", 1)
-- Tu Mapper 2: ("product_123", 1), ("product_789", 1)
-- Tu Mapper 3: ("product_123", 1)

-- Sau k-way merge (sorted):
key = "product_123" -> values = [1, 1, 1]
key = "product_456" -> values = [1]
key = "product_789" -> values = [1]

5. Phase Reduce — tổng hợp từng key

Sau khi Shuffle hoàn thành, reducer nhận một dòng sorted key-value. Hàm reduce được gọi cho mỗi key với toàn bộ danh sách value:

-- Job: dem so don hang theo san pham
function reduce(product_id, counts):
    total <- 0
    for each c in counts:
        total <- total + c
    emit(product_id, total)

Output của reducer được ghi ra HDFS — là derived data (từ bài trước): phái sinh, dựng lại được, không sửa input.

6. Vì sao Shuffle chiếm 1/3 thời gian job?

Theo các đo đạc trên workload thực tế, shuffle chiếm khoảng 1/3 tổng thời gian job điển hình. Nguyên nhân rõ ràng từ con số cụ thể:

Cuộc thi Daytona GraySort 2014, sort 100 TB dữ liệu tạo ra 500 TB disk I/O và 200 TB network I/O — tức mỗi byte dữ liệu cần khoảng 5 byte I/O. Phần lớn trong đó là Shuffle.

Tại sao nhiều I/O như vậy? Mỗi sorted spill của mapper ghi ra đĩa, được đọc lại khi merge, gửi qua mạng, ghi ra đĩa phía reducer, rồi đọc lại khi merge phía reducer. Một byte data đi qua nhiều lần đọc-ghi trước khi tới reducer.

Đòn bẩy giảm shuffle: compress map output trước khi ghi ra đĩa và gửi qua mạng giảm lượng data shuffle xuống khoảng 70% trong benchmark thực tế. CPU đánh đổi lấy I/O — thường có lợi vì mạng/đĩa chậm hơn CPU.

7. Pitfall — shuffle không miễn phí và hot key

-- order_line co 50 truong, ~10 KB moi dong
function map(order_line):
    emit(order_line.product_id, order_line)   -- 10 KB cho moi cap

-- 40 trieu dong * 10 KB = 400 GB phai shuffle qua mang

function map(order_line):
    emit(order_line.product_id, order_line.amount)  -- ~100 byte thay vi 10 KB
-- 40 trieu dong * 100 B = 4 GB -- giam 100 lan

-- MAP PHASE: them random suffix de trai data
function map(order_line):
    salt <- random(0, 9)                          -- 10 bucket
    new_key <- order_line.product_id + "_" + salt
    emit(new_key, order_line.amount)
-- "iphone-15_0", "iphone-15_1", ..., "iphone-15_9"
-- -> trai deu cho 10 reducer thay vi 1

-- REDUCE PHASE 1: tong hop tung bucket
-- Moi reducer chi nhan 1/10 data cua hot key

-- JOB 2 - gop lai:
function map2(new_key, subtotal):
    base_key <- strip_suffix(new_key)
    emit(base_key, subtotal)

function reduce2(product_id, subtotals):
    emit(product_id, sum(subtotals))

8. 📚 Deep Dive

9. Liên hệ các bài khác

• Bài 01 — Tư duy batch: MapReduce là dòng lệnh cut | sort | uniq -c phóng to ra cả cluster — map giống cut (biến đổi từng dòng), shuffle giống sort (gom theo key), reduce giống uniq -c (tổng hợp). Input bất biến + output phái sinh từ bài trước là nền để hiểu vì sao task re-execution an toàn.
• Bài 03 — Join trong batch: sort-merge join dùng lại đúng cơ chế shuffle đã học ở bài này — gom record cùng join key về một reducer, rồi merge hai luồng đã sort.
• External sort: Shuffle là external sort chạy phân tán — nếu muốn hiểu vì sao reducer merge k sorted run bằng min-heap, bài đó đào sâu cơ chế chunk-sort-merge.

10. Tóm tắt

• MapReduce chia job thành 3 pha: Map (phát key-value song song), Shuffle (gom + sort + phân phối đúng reducer), Reduce (tổng hợp). Lập trình viên chỉ viết Map và Reduce; framework tự lo Shuffle và fault tolerance.
• Shuffle = external sort phân tán: mapper sort cục bộ → spill sorted run ra đĩa → reducer fetch từ tất cả mapper → merge nhiều sorted run thành dòng sorted toàn cục. Kết quả: mọi value cùng key đến reducer liên tiếp nhau, đã sắp xếp.
• Shuffle chiếm khoảng 1/3 thời gian job điển hình (đo trên workload thực tế). Giảm data emit trong map (chỉ emit field cần), compress map output là hai đòn bẩy lớn nhất.
• Fault tolerance: task chết → chạy lại task đó từ input HDFS (bất biến). Không cần replica dữ liệu trung gian — tính lại từ đầu là đủ.
• Hot key (skew): hash chia đều key khác nhau, KHÔNG chia đều record. Hot key → một reducer nhận hàng triệu record → straggler. Fix: salting (thêm random suffix, cần job thứ hai để gộp).
• Combiner (tùy chọn): pre-aggregate cùng key ngay tại mapper trước shuffle — chỉ dùng được với hàm commutative và associative (tổng, max, min; không phải average hay median).

11. Tự kiểm tra

Bài tiếp theo: Join trong batch — sort-merge, broadcast hash, partitioned hash