Dataflow engine & fault tolerance — Spark vượt MapReduce

TL;DR: Dataflow engine (Spark, Tez) tổ chức công việc thành DAG (đồ thị có hướng không chu trình) — vertex là tác vụ, edge là luồng dữ liệu — thay vì chuỗi map → shuffle → reduce cứng của MapReduce. Điểm khoá: Spark sort 100 TB trong 23 phút, Hadoop cần 72 phút — cả hai chạy trên disk; Spark thắng vì pipelining (ít ghi–đọc đĩa hơn), không phải vì RAM. Hadoop materialize sau mỗi bước (N bước = 2N disk I/O); Spark chỉ ghi đĩa ở ranh giới shuffle, pipeline các bước còn lại. Khi node chết, Spark recompute từ lineage — đồ thị "ai sinh ra ai" — không cần replica trung gian. Nguyên tắc cốt lõi: input gốc bất biến → tính lại được.

Hãy nhìn một pipeline đơn giản: đọc log → lọc lỗi → ghép bảng user → đếm lỗi mỗi user. Trong MapReduce (bài 02), đây là ba job độc lập chạy nối tiếp. Sau mỗi job, kết quả ghi xuống HDFS; job sau đọc lên từ HDFS. Ba bước = sáu lần đọc/ghi đĩa, trong đó bốn lần là chi phí thuần tuý của cơ chế fault tolerance — không đóng góp gì cho kết quả.

Năm 2014, Spark sort 100 TB trong 23 phút với 206 nút, Hadoop cần 72 phút với 2.100 nút — nhanh hơn 3×, ít máy hơn 10×. Cả hai chạy hoàn toàn trên disk. Spark thắng vì DAG + pipelining giúp cắt hầu hết các lần ghi–đọc đĩa trung gian không cần thiết. Bài này giải thích cơ chế đằng sau khoảng cách đó và vì sao fault tolerance theo kiểu lineage không cần replica.

1. Analogy — băng chuyền vs bồn trữ trung gian

Hình dung nhà máy bia với 5 bước: nghiền mạch → nấu → lên men → lọc → đóng chai.

Kiểu A (materialized): sau mỗi bước, toàn bộ bán thành phẩm đổ vào một bồn trữ trung gian. Bước tiếp theo chỉ bắt đầu khi múc từ bồn đó lên. Năm bước = bốn bồn trữ = bốn lần rót-và-múc. Khi lò nấu hỏng, bồn sau lò vẫn còn nguyên — khởi động lại từ bồn đó, không cần nấu lại từ đầu.

Kiểu B (pipelined): băng chuyền liên tục — đầu ra lò nấu chảy thẳng vào nồi lên men, không qua bồn trữ. Nhanh hơn vì bỏ được các lần rót-và-múc. Khi lò nấu hỏng, không có bồn trữ giữa chừng — phải nấu lại từ nguyên liệu gốc theo công thức (recipe) đã lưu sẵn.

2. DAG là gì — từ "pha cứng" sang đồ thị

MapReduce ép mọi bài toán vào khuôn map → shuffle → reduce cứng. Nhiều bài toán tự nhiên hơn khi biểu diễn thành chuỗi bước tuỳ ý — lọc, chiếu, gộp, join, sort — không nhất thiết nhét vừa hai pha.

Dataflow engine tổ chức công việc thành DAG (Directed Acyclic Graph — đồ thị có hướng không chu trình):

• Vertex = một tác vụ tính toán (filter, map, join, aggregate, sort).
• Edge = luồng dữ liệu từ vertex trước sang vertex sau.
• Không chu trình (acyclic) — không có vòng phụ thuộc vòng quanh.

Ví dụ pipeline "đọc log → lọc lỗi → lấy user_id → shuffle → đếm lỗi mỗi user":

flowchart LR
  A["Read log"] --> B["Filter ERROR"]
  B --> C["Map: user_id"]
  C --> D["Shuffle by user_id"]
  D --> E["Count per user"]
  E --> F["Write output"]

Spark phân tích DAG này và chia thành stage (giai đoạn). Ranh giới stage = chỗ cần shuffle (dữ liệu phải phân phối lại qua mạng). Trong ví dụ: Stage 1 gồm Read → Filter → Map; Stage 2 gồm Count → Write; giữa hai stage là Shuffle by user_id — bắt buộc ghi đĩa để các task sau đọc từ nhiều máy khác nhau.

Nhờ DAG, engine thấy toàn bộ kế hoạch trước khi chạy: ghép các bước liền nhau thành một pipeline, tránh ghi đĩa không cần thiết, chọn đúng số partition cho từng stage.

3. Vì sao MapReduce nhiều bước chậm?

Giả sử pipeline có ba bước: lọc → map → tính tổng.

MapReduce: mỗi bước là một job độc lập. Sau mỗi job, kết quả ghi xuống HDFS; job tiếp theo đọc lên từ HDFS.

Job 1:  Read input  --> Map --> Shuffle --> Reduce --> WRITE HDFS
Job 2:  READ HDFS   --> Map --> Shuffle --> Reduce --> WRITE HDFS
Job 3:  READ HDFS   --> Map --> Shuffle --> Reduce --> Write output

Tong disk I/O: 3 ghi + 3 doc = 6 lan (4 lan trung gian khong can thiet)

Bốn lần ghi/đọc trung gian tồn tại chỉ để cung cấp điểm khởi động lại khi lỗi — job tiếp theo đọc từ HDFS thay vì từ output của job trước đang chạy trên bộ nhớ. Với pipeline năm bước, con số là 10 lần đọc/ghi; với ML iterative lặp lại 20 vòng, số lần nhân lên theo số vòng lặp.

Đây không phải lỗi thiết kế — đây là đánh đổi có chủ đích của MapReduce: fault tolerance đơn giản đổi lấy disk I/O tốn kém. Dataflow engine (Spark, Tez) chọn điểm đánh đổi khác.

4. Materialization vs pipelining — tại sao Spark nhanh hơn?

Materialization là ghi kết quả trung gian ra đĩa sau mỗi bước (Hadoop: ghi sau mỗi reduce).

Pipelining là luồng dữ liệu chạy trực tiếp từ operator này sang operator kế tiếp, không ghi đĩa giữa chừng.

Spark áp dụng pipelining trong mỗi stage: mọi operator từ đầu stage đến cuối stage chạy liên tục trên cùng partition, không ghi đĩa trung gian. Chỉ ở ranh giới shuffle (stage boundary) mới ghi đĩa — vì shuffle cần phân phối lại dữ liệu qua mạng, mapper phải ghi output trước khi reducer ở máy khác đọc vào.

MapReduce 3 buoc (materialized):
  [Read --> Map] --> WRITE HDFS --> READ HDFS --> [Shuffle --> Reduce] --> WRITE HDFS
  --> READ HDFS --> [Map --> Reduce] --> Write output
  Disk I/O: 2 ghi trung gian + 2 doc trung gian = 4 lan khong can thiet

Spark 3 buoc (pipelined):
  Stage 1: [Read --> Filter --> Map] --> WRITE shuffle boundary
  Stage 2: READ shuffle --> [Aggregate --> Write output]
  Disk I/O: 1 ghi + 1 doc = 2 lan (chi o shuffle, bat buoc)

Daytona GraySort 2014 (số liệu đã kiểm chứng): sort 100 TB, cả hai hệ chạy trên disk, không cache RAM:

Spark nhanh hơn 3× với ít hơn 10× máy — do pipelining giảm disk I/O, không phải do RAM. Đây là hiểu lầm phổ biến nhất: "Spark nhanh vì giữ dữ liệu trong bộ nhớ". Trong benchmark chuẩn 100 TB (lớn hơn nhiều tổng RAM của 206 nút), Spark chạy hoàn toàn trên disk và vẫn thắng 3×. Cache RAM là tính năng phụ trợ cho ML iterative (cùng RDD tính nhiều lần), không phải lý do nền tảng.

Đáp án: Stage boundary nằm ngay trước Shuffle. Stage 1 gồm Read → Filter → Map — ba operator này chạy pipelined trên cùng partition, không ghi đĩa trung gian. Operator Shuffle bắt buộc ghi đĩa vì nó cần phân phối lại dữ liệu qua mạng: mapper (Stage 1) ghi partition output ra đĩa cục bộ theo key; reducer (Stage 2) fetch qua mạng rồi đọc lên. Stage 2 gồm Aggregate — chạy pipelined sau khi shuffle hoàn tất.

5. Fault tolerance bằng lineage — không cần replica

Khi node chứa dữ liệu trung gian chết, cần khôi phục bằng cách nào?

Hadoop approach: vì mọi bước đã ghi ra HDFS (replicated 3×), chỉ cần restart task đó trên node khác, đọc input từ replica. Đơn giản, nhưng đổi lấy disk I/O cho toàn bộ kết quả trung gian.

Spark approach: Spark không lưu replica dữ liệu trung gian. Thay vào đó, nó lưu lineage — đồ thị mô tả mỗi partition được sinh ra như thế nào từ partition upstream qua phép biến đổi nào. Khi partition mất, Spark tra lineage và tính lại đúng partition đó từ input upstream.

Đây là áp dụng trực tiếp nguyên tắc từ bài 01 — Tư duy batch: input bất biến → tính lại được. Input gốc trên HDFS không thay đổi, nên recompute luôn cho kết quả đúng.

Ví dụ lineage cho pipeline lọc → map → shuffle → đếm:

RDD_A (raw log, tren HDFS -- bat bien)
  |-- filter(severity=ERROR)  --> RDD_B (error records)
       |-- map(to user_id)    --> RDD_C (user_id stream)
            |-- shuffle       --> RDD_D (partitioned by user_id)
                 |-- count()  --> RDD_E (count per user)

Giả sử partition 4 của RDD_D mất (node chứa nó chết):

Đáp án: Spark tra lineage ngược: RDD_D sinh từ RDD_C qua shuffle; RDD_C sinh từ RDD_B qua map; RDD_B sinh từ RDD_A (input gốc, còn nguyên trên HDFS) qua filter. Spark recompute chỉ các partition của RDD_B và RDD_C cần thiết để tạo lại partition 4 của RDD_D — các partition khác không bị đụng. Sau đó shuffle lại phần data đó, sinh lại partition 4 của RDD_D, rồi tiếp tục tính RDD_E.

Trade-off — checkpoint khi lineage dài: pipeline ML iterative 20 vòng lặp tích lũy lineage qua mỗi vòng. Recompute một partition ở cuối vòng 19 có thể kéo theo tính lại từ đầu toàn bộ 20 vòng — đắt hơn lưu replica. Giải pháp: checkpoint() ghi snapshot một RDD ra HDFS tại điểm giữa, cắt lineage tại đó. Từ đây về sau, recompute chỉ cần từ checkpoint.

6. Pitfall tổng hợp

❌ Nhầm 1 — "Spark nhanh vì RAM": Trong Daytona GraySort 2014 (100 TB, benchmark chuẩn), cả hai hệ chạy trên disk. Spark thắng nhờ pipelining (ít disk I/O), không phải nhờ RAM.

✅ Cache RAM (df.cache(), rdd.persist()) hữu ích khi cùng một RDD được tính nhiều lần (ML iterative, nhiều action trên cùng DataFrame) — nhưng đó là tối ưu phụ trợ, không phải lý do nền tảng. Dùng persist() tràn lan gây OOM và không giúp pipeline chạy một lần.

❌ Nhầm 2 — "Spark không bao giờ ghi đĩa trung gian": Spark vẫn ghi đĩa tại mỗi shuffle boundary — bắt buộc, không thể tránh.

✅ Số lần ghi đĩa = số lần shuffle trong DAG, không phải số operator. DAG 10 operator nhưng 1 shuffle = ghi đĩa 1 lần. "Ít hơn Hadoop" không có nghĩa là "không bao giờ".

❌ Nhầm 3 — "Lineage thay thế hoàn toàn checkpoint": Lineage dài (ML iterative 20 vòng) làm recompute đắt hơn lưu replica.

✅ Dùng checkpoint() (ghi HDFS replicated, cắt lineage) sau mỗi 5–10 bước hoặc mỗi vòng iterative. Shuffle boundary tuy có ghi đĩa nhưng chỉ là file cục bộ, ephemeral — mất khi node chết nên KHÔNG thay được checkpoint durable; nó chỉ giúp giảm recompute khi retry trong cùng stage.

7. 📚 Deep Dive — RDD và dataflow engine

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

• Bài 01 — Tư duy batch: lineage recompute là ứng dụng trực tiếp "input bất biến → tính lại được" — Spark giữ input gốc read-only nên luôn recompute được khi mất dữ liệu trung gian.
• Bài 02 — MapReduce: MapReduce là trường hợp đặc biệt của dataflow với DAG chỉ có hai pha cứng (map + shuffle + reduce). Dataflow engine tổng quát hoá lên DAG nhiều stage tuỳ ý và bỏ giới hạn hai pha.
• Bài 03 — Join trong batch: sort-merge join trong Spark là ứng dụng của stage DAG — bước sort nằm trong Stage 1, bước merge nằm trong Stage 2, ranh giới là shuffle redistribute data theo join key.
• Bài 05 — Thiết kế batch pipeline: lab tổng hợp — bạn tự thiết kế DAG nhiều stage và lập luận khi nào dùng checkpoint() để cắt lineage.

9. Tóm tắt

• DAG (Directed Acyclic Graph): vertex = tác vụ, edge = luồng dữ liệu. Dataflow engine tổng quát hoá MapReduce — pipeline tuỳ ý thay vì map → shuffle → reduce cứng.
• Stage boundary = chỗ shuffle — nơi duy nhất bắt buộc ghi đĩa. Các operator trong cùng stage chạy pipelined, không ghi đĩa trung gian.
• Materialization (Hadoop): ghi đĩa sau mỗi bước → N bước = 2N disk I/O. Pipelining (Spark): ghi đĩa chỉ ở shuffle boundary.
• Spark thắng Daytona 2014 (100 TB, 23 phút, cả hai trên disk) nhờ ít disk I/O, không phải nhờ RAM. Dùng persist() sai chỗ không giúp gì nếu bottleneck là shuffle I/O.
• Fault tolerance bằng lineage recompute: Spark lưu "công thức" sinh partition, không lưu replica trung gian. Node chết → tính lại partition đó từ input upstream theo lineage.
• Checkpoint cắt lineage dài: cần thiết khi ML iterative hoặc pipeline vượt 5–10 bước; ghi snapshot RDD ra HDFS tại điểm giữa để giới hạn phạm vi recompute.

10. Tự kiểm tra

Bài tiếp theo: Thiết kế batch pipeline nhiều bước