Spark Execution Model

From Programming Model to Execution Model

Ở Spark Programming Model, mình đứng ở phía user: chọn một API để nhìn data, nối các operations lại với nhau, rồi mô tả data cần được transform như thế nào. Ở mức đó, đoạn code như filter, join, groupBy, count vẫn chỉ là một description of computation. Nó nói logic cần làm gì, nhưng chưa nói cụ thể cluster sẽ chạy logic đó ra sao.

Execution model bắt đầu từ câu hỏi kế tiếp: khi Spark đã có description này, nó biến description đó thành distributed work như thế nào?

Cluster không thể chạy trực tiếp một chuỗi method calls như cách user đọc code từ trên xuống dưới. Spark cần giữ lại computation trong một representation mà framework hiểu được, phân tích dependency giữa các phần data, xác định phần nào có thể chạy local, phần nào cần data movement, rồi chia toàn bộ computation thành các đơn vị work đủ nhỏ để schedule trên cluster.

Nói ngắn gọn, programming model trả lời:

User mô tả computation bằng abstraction nào?

Execution model trả lời:

Spark dùng description đó để lập plan, chia nhỏ work, và chạy song song như thế nào?

Computation Before Execution

Spark không thực thi từng API call ngay tại thời điểm user gọi nó. Khi user viết filter, select, join hay groupBy, Spark thường chỉ đang mở rộng description của computation. Chỉ khi có action yêu cầu kết quả, Spark mới cần quyết định toàn bộ chuỗi đó phải được chạy ra sao.

Tùy API, representation bên trong có thể được nghĩ hơi khác nhau:

DataFrame / Dataset / SQL
  -> query plan

RDD
  -> lineage / dependency graph

Với DataFrame, Dataset và Spark SQL, Spark có schema và expressions nên có thể giữ computation dưới dạng query plan . Query plan cho Spark biết cột nào được đọc, expression nào được tính, filter/join/aggregate nào xuất hiện, và output cần có shape gì.

Với RDD, Spark giữ computation dưới dạng lineage hoặc dependency graph. Lineage cho Spark biết RDD hiện tại được tạo ra từ RDD nào, bằng operation gì, và partition nào của RDD này phụ thuộc vào partition nào của RDD trước.

Hai đường này khác nhau về mức structure mà Spark nhìn thấy, nhưng cùng phục vụ một mục đích: Spark cần nhìn computation như một graph/plan trước khi biến nó thành execution thật.

Partitions: The Unit of Parallel Data

Spark có thể chạy song song vì distributed data không được xử lý như một khối duy nhất. Data được chia thành nhiều partitions . Mỗi partition là một phần của dataset, và trong execution, Spark có thể giao work trên từng partition cho các task khác nhau.

Ví dụ một dataset có thể được chia thành ba partitions:

dataset
  ├─ partition 1
  ├─ partition 2
  └─ partition 3

Nếu operation chỉ cần xử lý từng partition độc lập, Spark có thể chạy nhiều partition song song. Đây là cầu nối đầu tiên giữa data abstraction và distributed execution: user nhìn thấy một DataFrame/RDD/table, còn Spark nhìn thấy nhiều partitions có thể được xử lý bởi nhiều tasks.

Điểm quan trọng là partition không chỉ là cách chia data để lưu trữ. Trong execution model, partition còn quyết định độ song song tự nhiên của computation. Một stage có bao nhiêu tasks thường gắn với việc nó cần xử lý bao nhiêu partitions ở bước đó.

Dependencies Shape Execution

Sau khi nhìn data theo partitions, Spark cần hiểu dependency giữa các partitions. Không phải operation nào cũng tạo ra cùng một dependency shape.

Một số transformations có dependency hẹp. Ví dụ filter, select hoặc map thường có thể xử lý từng partition riêng lẻ: output partition này chủ yếu được tính từ một input partition tương ứng. Spark có thể pipeline các operation này lại với nhau vì data không cần rời khỏi partition hiện tại.

Một số transformations có dependency rộng. Ví dụ groupBy, orderBy, distinct hoặc nhiều dạng join cần gom những record liên quan từ nhiều partitions khác nhau. Lúc này một output partition có thể phụ thuộc vào data đến từ nhiều input partitions. Spark không thể chỉ xử lý cục bộ từng partition nữa; nó phải phân phối lại data theo một partitioning mới.

Đó là lý do dependency shape quan trọng:

partition dependency
  -> cho biết work nào có thể chạy local
  -> cho biết chỗ nào phải di chuyển data
  -> giúp Spark xác định boundary giữa các vùng execution

Ở programming model, user chỉ thấy mình gọi groupBy("city"). Ở execution model, Spark phải hiểu rằng những record cùng city có thể đang nằm rải rác ở nhiều partitions khác nhau, nên muốn count đúng thì cần đưa chúng về cùng một grouping/partitioning phù hợp.

Shuffle as an Execution Boundary

Khi dependency rộng buộc data từ nhiều partitions phải được gom hoặc phân phối lại, Spark thường cần shuffle .

Shuffle là một boundary lớn trong execution. Trước shuffle, mỗi task có thể xử lý phần data local của nó. Sau đó intermediate data được partition lại theo key hoặc rule mới, ghi/đọc qua shuffle layer, rồi các tasks phía sau mới tiếp tục xử lý trên data đã được phân phối lại.

Ví dụ với groupBy("city").count(), mỗi partition ban đầu có thể chứa events của nhiều city. Để tính tổng count theo city, Spark cần đảm bảo record cùng city được gom về đúng nơi xử lý. Điều này thường tạo ra shuffle: data được chia lại theo key city, rồi stage sau shuffle mới aggregate kết quả cuối cùng.

before shuffle
  partition 1: VN/HN, VN/HCM, SG/Singapore
  partition 2: VN/HN, TH/Bangkok
  partition 3: VN/HCM, VN/HN

shuffle by city
  -> all HN records together
  -> all HCM records together
  -> all Singapore/Bangkok records together

Shuffle đắt vì nó đưa data ra khỏi phạm vi xử lý local: phải serialize, ghi intermediate data, truyền qua network hoặc đọc từ shuffle storage, rồi deserialize để xử lý tiếp. Vì vậy Spark cố pipeline những operation có thể chạy liền nhau, và chỉ tạo boundary lớn khi dependency của computation thật sự yêu cầu.

Jobs, Stages, and Tasks

Khi action yêu cầu kết quả, Spark materialize computation thành các execution units: job, stage, và task.

Một action như count() có thể tạo ra một job. Job đó được chia thành stages dựa trên dependency và shuffle boundary. Trong mỗi stage, Spark tạo nhiều tasks để xử lý các partitions tương ứng.

action
  -> job
      -> stage 1
          -> task per partition
      -> shuffle boundary
      -> stage 2
          -> task per partition

Nếu computation chỉ gồm các narrow transformations như read -> filter -> select -> count, Spark có thể pipeline nhiều bước trong cùng một stage vì mỗi partition được xử lý khá độc lập. Nếu computation có wide transformation như groupBy hoặc join, Spark thường phải cắt job thành nhiều stages quanh shuffle boundary.

Điểm cần nhớ không phải là đếm chính xác mọi job/stage/task bằng mắt từ code. Điểm quan trọng hơn là causal chain:

operation dependency shape
  -> shuffle boundary or not
  -> stage boundary or not
  -> tasks over partitions

Reading a Simple Example

Quay lại một flow quen thuộc:

result = events \
    .filter(events.country == "VN") \
    .join(users, "user_id") \
    .groupBy("city") \
    .count()

result.show()

Ở programming model, đây là một data transformation pipeline: lọc events ở Việt Nam, join với users, group theo city, rồi count.

Ở execution model, Spark đọc flow này bằng câu hỏi khác:

• filter có thể chạy local trên từng partition không?
• join có cần gom records cùng user_id về cùng partition không?
• groupBy("city") có tạo shuffle mới không?
• Sau mỗi shuffle, stage boundary nằm ở đâu?
• Mỗi stage sẽ có bao nhiêu tasks dựa trên số partitions ở thời điểm đó?

Cùng một đoạn code, nhưng hai page nhìn nó ở hai tầng khác nhau. Programming model nói user mô tả logic bằng abstraction nào. Execution model nói Spark dùng structure đó để tạo ra work graph chạy được trên cluster.

My Summary

Từ góc nhìn cá nhân, Spark execution model là bước dịch từ description of computation sang distributed work. Page trước dừng ở chỗ user mô tả data transformation bằng RDD/DataFrame/Dataset/SQL. Page này đi tiếp một tầng: Spark giữ description đó thành query plan hoặc lineage, nhìn data theo partitions, phân tích dependency giữa partitions, rồi quyết định chỗ nào có thể pipeline và chỗ nào phải tạo shuffle boundary.

Ba khái niệm quan trọng nhất ở tầng này là partition, dependency và shuffle. Partition cho Spark đơn vị data để chạy song song. Dependency cho Spark biết output partition phụ thuộc input partition nào. Shuffle đánh dấu những chỗ computation không thể tiếp tục chạy local mà phải phân phối lại data trên cluster.

Từ đó, job/stage/task chỉ là cách Spark đóng gói work để schedule: action tạo job, job được chia thành stages quanh shuffle boundaries, và mỗi stage gồm các tasks xử lý partitions. Driver/executor là runtime structure chạy những work đó, nhưng đó là câu hỏi của page sau.