Điều gì thực sự xảy ra khi bạn chạy một tác vụ Spark?

Giới thiệu

Khi tập dữ liệu ngày càng lớn, việc xử lý chúng hiệu quả trên một máy duy nhất có thể trở nên khó khăn do giới hạn về năng lực CPU, bộ nhớ và thông lượng lưu trữ.

Apache Spark giải quyết những thách thức này bằng cách cung cấp một công cụ xử lý dữ liệu phân tán, có thể chia khối lượng công việc thành các phân vùng và thực thi tác vụ song song trên các lõi CPU cũng như nhiều máy.

Tuy nhiên, Spark không chỉ đơn giản là bổ sung thêm máy.

Hiểu cách các thành phần phối hợp với nhau—Driver, Executor, RDD, DataFrame, phân vùng, Job, Stage và Task—là điều thiết yếu để thiết kế các quy trình dữ liệu hiệu quả và đáng tin cậy.

Trong bài viết này, tôi sẽ trình bày kiến trúc, cơ chế thực thi và những yếu tố hiệu năng chính của Spark thông qua ví dụ thực tế về tổng hợp doanh số thương mại điện tử.

1. Vì sao chọn Apache Spark – xử lý dữ liệu song song

Apache Spark là một công cụ tính toán phân tán được thiết kế để xử lý dữ liệu quy mô lớn và phục vụ kỹ thuật dữ liệu.

Spark xử lý dữ liệu lớn bằng phương pháp xử lý dữ liệu song song:

Hình A. Kiến trúc tổng quan của cụm phân tán Apache Spark (cụm, nút điều phối và nút công nhân), so sánh với việc thực thi trên một nút (Do Kuriko IWAI tạo)

So với xử lý trên một nút (bên phải, Hình A), Spark chia công việc thành các tác vụ bằng cách sử dụng nhiều nút và thực thi chúng trên các tài nguyên sẵn có ở nhiều máy.

Ở đây, phân tán không có nghĩa là bắt buộc phải dùng nhiều máy. Spark có thể chạy ở chế độ cục bộ trên một máy duy nhất. Ngay cả trong trường hợp đó, dữ liệu vẫn có thể được chia thành các phân vùng và xử lý song song.

2. Kiến trúc cốt lõi của Apache Spark

Spark gồm một số thành phần chính:

• Tiến trình điều phối Spark (driver): Tiến trình trung tâm điều phối ứng dụng Spark.
• Trình quản lý cụm: Dịch vụ bên ngoài chịu trách nhiệm phân bổ tài nguyên phần cứng như lõi CPU và bộ nhớ.
• Các nút công nhân (worker): Máy vật lý hoặc máy ảo, hay nút tính toán trong cụm.
• Các tiến trình thực thi: Các tiến trình công nhân phân tán được khởi chạy trên các nút công nhân.
• Java Virtual Machine (JVM): Môi trường thời gian chạy cần thiết để thực thi mã byte của Spark (vì Spark được viết bằng Scala/Java).

Trước tiên, driver điều phối một ứng dụng Spark và chuyển ứng dụng đó thành một đồ thị có hướng không chu trình (DAG) gồm các tác vụ thực thi vật lý, sau đó lập lịch cho chúng trên các executor.

Sau đó, trình quản lý cụm phân bổ tài nguyên theo hệ thống quản lý tài nguyên của môi trường triển khai.

Các nút công nhân cung cấp tài nguyên tính toán cho các executor; các executor thực thi tác vụ và lưu giữ dữ liệu trong quá trình thực thi.

Một số môi trường được quản lý như Databricks sử dụng một executor cho mỗi worker, nhưng đây không phải quy tắc chung của Spark. Tùy theo cấu hình cụm, một worker có thể chứa nhiều executor.

2-1. Trường hợp thương mại điện tử – cấu trúc dữ liệu đầu vào

Hãy hình dung một công ty thương mại điện tử nhỏ chuyên bán hàng tuyển chọn.

Công ty có 10,000 bản ghi doanh số, mỗi bản ghi gồm một mã hàng tồn kho (SKU) và số lượng bán ra. Giả sử có ba SKU khả dĩ: A, B, và C.

Bảng 1. Ví dụ về phân bổ bản ghi doanh số thương mại điện tử và các phép tổng hợp dự kiến.

Mục tiêu của chúng ta là tính tổng số đơn vị hợp lệ đã bán theo từng SKU:

1. Lọc bỏ các bản ghi mà qty <= 0, rồi
2. Tổng hợp các mẫu theo SKU.

Trong bộ dữ liệu đầy đủ dùng để minh họa, giả sử sau khi lọc còn lại 9,800 trong tổng số 10,000 bản ghi. Kết quả cuối cùng mà Spark dự kiến trả về là:

[("A", 8000), ("B", 10500), ("C", 9200)]

2-2. Cụm Spark

Tôi sẽ xác định một cụm đơn giản hóa cho trường hợp sử dụng này.

Các nút công nhân (worker)

Tôi sẽ thiết lập 3 nút công nhân, mỗi nút có loại phiên bản gồm 8 vCPU và 32 GiB RAM:

Hình B. Phân bổ tài nguyên trên các nút công nhân (Do Kuriko IWAI tạo)

Tổng cộng, hệ thống có 24 vCPU.

Các executor của Spark và phân bổ CPU/bộ nhớ

Giả sử tôi sẽ cấp:

• 7 executor.
• 2 lõi vCPU cho mỗi executor.
• 3 GiB bộ nhớ heap JVM cho mỗi executor.

Ứng dụng Spark yêu cầu trình quản lý cụm phân bổ tài nguyên cho 7 executor này.

Sau đó, trình quản lý cụm quyết định cách phân bổ CPU và bộ nhớ dựa trên mức sử dụng tài nguyên của từng worker.

Chẳng hạn, trình quản lý quyết định tạo 14 ô tác vụ — 2 lõi vCPU cho mỗi executor × 7 executor:

Hình C-1. Lõi vCPU, bộ nhớ và phân bổ executor giữa các worker khi tài nguyên ở Worker A bị mất cân bằng

Lưu ý rằng 3 GiB bộ nhớ được phân bổ cho Executor không phải là tổng dung lượng bộ nhớ của nó, vì cần dành thêm bộ nhớ dự phòng cho các worker và những tiến trình thực thi khác.

Giả sử một tác vụ cần 1 vCPU (spark.task.cpus = 1), mỗi worker phân bổ vCPU của mình:

Hình C-2. Phân bổ vCPU song song giữa các worker

Điều này cho thấy hai điểm chính:

1. Mỗi worker có tỷ lệ sử dụng CPU riêng (Hình C-2 cho thấy mức sử dụng của Worker A là 75%, trong khi Worker B và C vẫn ở mức 50%).
2. Mặc dù hệ thống có 24 lõi vCPU (8 vCPU/worker × 3 worker), nhưng chỉ có thể sử dụng 14 lõi vCPU; 10 lõi còn lại không được dùng cho các tác vụ này.

2-3. Máy ảo Java (JVM) phía sau Spark

Apache Spark được xây dựng trên Máy ảo Java (JVM).

JVM là môi trường chạy cần thiết để thực thi mã bytecode của Spark.

Vì Spark được viết bằng Scala/Java:

• Khi sử dụng Scala hoặc Java trong ứng dụng: Mã chạy trực tiếp bên trong JVM.
• Khi sử dụng PySpark (Python): Một tiến trình Python chạy mã, nhưng giao tiếp với trình điều khiển dựa trên JVM thông qua một cầu nối như Py4J. Các executor Spark trên các nút worker vẫn chạy bên trong JVM; JVM sau đó quản lý các worker Python để thực thi mã Python và các hàm do người dùng định nghĩa bằng Python (Python UDF).

Trong cả hai trường hợp, JVM quản lý các luồng thực thi, việc phân bổ bộ nhớ và quá trình thu gom rác cho cả trình điều khiển lẫn các executor.

Các JVM của trình điều khiển và executor trong ứng dụng PySpark

Trong ứng dụng PySpark, trình điều khiển Python giao tiếp với trình điều khiển Spark chạy trong JVM;

• Trình điều khiển Python: Nơi viết và thực thi mã PySpark ở cấp ứng dụng.
• JVM của trình điều khiển: Điều phối việc thực thi và lập lịch Spark.
• JVM của executor: Chạy các tác vụ Spark và quản lý tài nguyên thực thi. Các tiến trình worker Python tham gia khi tác vụ cần thực thi bằng Python.
• JVM của máy khách (tiến trình máy khách hoặc máy gửi tác vụ): JVM chạy trên máy cục bộ hoặc nút biên/cổng.

Như hình minh họa, JVM của máy khách trước tiên kết nối với JVM của trình điều khiển, sau đó JVM này điều phối với các JVM của executor:

Hình D. Mối quan hệ tổng thể giữa JVM của máy khách, trình điều khiển và executor

Từ đây, JVM của trình điều khiển lưu trữ Spark Driver để quản lý ứng dụng, còn các JVM của executor lưu trữ từng executor riêng lẻ, chịu trách nhiệm thực thi các tác vụ đó.

Mỗi JVM của trình điều khiển đều có bộ nhớ heap JVM để xử lý dữ liệu, lưu vào bộ nhớ đệm và quản lý các đối tượng do người dùng tạo một cách hiệu quả.

2-4. Tìm hiểu chuyên sâu về tập dữ liệu phân tán có khả năng phục hồi (RDD)

Tập dữ liệu phân tán có khả năng phục hồi (RDD) là một tập hợp các phần tử bất biến, được chia thành các phân vùng mà Spark có thể xử lý song song.

RDD theo dõi các mối phụ thuộc giữa những phép biến đổi, cho phép Spark tính toán lại các phân vùng bị mất trong nhiều tình huống lỗi.

Tên gọi này thể hiện ba đặc tính:

• Có khả năng phục hồi: Spark ghi lại dòng dữ liệu — RDD nào phụ thuộc vào RDD trước đó — và thường có thể tái tạo các phân vùng bị thiếu thay vì duy trì các bản sao của mọi kết quả trung gian.
• Phân tán: Tập dữ liệu được biểu diễn thành các phân vùng mà Spark có thể xử lý trên những executor khác nhau. Không bắt buộc phải có nhiều máy vật lý; chế độ cục bộ cũng hỗ trợ RDD.
• Tập dữ liệu: RDD có thể chứa các cặp khóa-giá trị, số nguyên, chuỗi, bộ giá trị và các phần tử được hỗ trợ khác — không chỉ các hàng quan hệ. API RDD có các thao tác như map, filter, reduceByKey, join, và aggregateByKey.

RDD là một mô hình trừu tượng về tập hợp logic và phép tính, chứ không phải tuyên bố rằng các phân vùng cụ thể được gán cố định cho các worker cụ thể.

Các phân vùng

Phân vùng là một phần dữ liệu logic và thường tương ứng với một tác vụ trong một giai đoạn.

Giả sử chúng ta tạo 100 phân vùng (100 phân vùng ban đầu này là lựa chọn có chủ đích để minh họa).

Với 10.000 phần tử đầu vào, mỗi phân vùng sẽ có khoảng 100 phần tử nếu dữ liệu đầu vào được phân phối tương đối đồng đều:

Hình E. Phân bổ phân vùng song song trên 100 phần dữ liệu ban đầu

3. API RDD và API DataFrame: Tối ưu hóa nội bộ bằng Catalyst

Ví dụ ban đầu về RDD giúp hiểu rõ quan hệ dòng dõi, các phân vùng và thao tác xáo trộn dữ liệu.

Tuy nhiên, trong các quy trình ETL có cấu trúc thường gặp, API DataFrame là lựa chọn khởi đầu được ưu tiên vì API này cung cấp các cột, thông tin lược đồ, biểu thức tương tự SQL và các thao tác mà trình tối ưu hóa có thể nhận biết.

Sơ đồ dưới đây minh họa mối quan hệ giữa các khái niệm:

Hình F. Mối quan hệ giữa các khái niệm đồ thị của Spark: quan hệ dòng dõi RDD, DataFrame và DAG thực thi của Spark

Cũng như RDD có quan hệ dòng dõi để xác định các phép biến đổi, kế hoạch truy vấn của DataFrame thể hiện truy vấn có cấu trúc của nó.

Spark xây dựng các tác vụ thực thi dựa trên các quan hệ phụ thuộc và toán tử vật lý.

Việc thực thi DataFrame có thể sử dụng các cơ chế dựa trên RDD ở bên trong, nhưng điều đó không có nghĩa là người dùng cần tự xây dựng quan hệ dòng dõi RDD khi viết mã DataFrame.

3-1. Kế hoạch truy vấn DataFrame

Kế hoạch truy vấn DataFrame là chuỗi các bước logic và vật lý do công cụ thực thi tạo ra (ô màu tím trong Hình L).

Truy vấn DataFrame trải qua một số dạng biểu diễn:

1. Kế hoạch logic: Mô tả các thao tác được yêu cầu (quét, lọc, nhóm, tổng hợp).
2. Kế hoạch logic đã phân tích và tối ưu hóa: Spark phân giải các tham chiếu và áp dụng những tối ưu hóa logic phù hợp, chẳng hạn như loại bỏ cột hoặc đẩy điều kiện lọc xuống.
3. Kế hoạch vật lý: Spark chọn các toán tử có thể thực thi, chẳng hạn như quét, tổng hợp từng phần/cuối cùng, trao đổi dữ liệu, kết nối và sắp xếp.
4. Thực thi: Kế hoạch vật lý được chạy bằng các cơ chế thực thi của Spark và cuối cùng bao gồm các tác vụ trên các executor.

Trong trường hợp sử dụng của chúng ta, kế hoạch vật lý có thể bao gồm các bước sau:

• Tổng hợp băm từng phần,
• Trao đổi dữ liệu theo SKU, và
• Tổng hợp băm cuối cùng.

3-2. So sánh các thao tác: API RDD và API DataFrame

Hãy so sánh API RDD với API DataFrame, lấy thao tác tổng hợp SKU làm ví dụ.

API RDD

Các phần tử là (sku, qty) các cặp. Logic tổng hợp bằng Python của chúng ta được nêu tường minh, và Spark tuân theo các quan hệ phụ thuộc của RDD:

result_rdd = (
    raw_rdd
    .filter(lambda row: row[1] > 0)
    .reduceByKey(lambda a, b: a + b, numPartitions=4)
)

rdd_results = result_rdd.collect()

# Example:
# [('A', 8000), ('B', 10500), ('C', 9200)]

API DataFrame

Giả sử sales_df có các cột sku (kiểu chuỗi) và qty (kiểu số):

from pyspark.sql import functions as F

result_df = (
    sales_df
    .filter(F.col("qty") > 0)
    .groupBy("sku")
    .agg(F.sum("qty").alias("total_qty"))
)

df_results = result_df.collect()

# Example:
# [Row(sku='A', total_qty=8000),
#  Row(sku='B', total_qty=10500),
#  Row(sku='C', total_qty=9200)]

Ở đây, cả hai ví dụ đều giải quyết cùng một vấn đề kinh doanh:

Lọc các bản ghi bán hàng không hợp lệ và tổng hợp số lượng hợp lệ theo SKU.

Và trong cả hai trường hợp, collect() sẽ kích hoạt quá trình thực thi của Spark và trả toàn bộ bản ghi kết quả về Driver.

Điểm khác biệt chính ở đây là cách biểu diễn kết quả:

• API RDD: Trả về các phần tử Python, chẳng hạn như (sku, total_qty) các bộ giá trị.
• API DataFrame: Trả về các đối tượng Row có cấu trúc với các trường được đặt tên, chẳng hạn như sku và total_qty.

Dưới đây là bảng so sánh tổng quan giữa API RDD và API DataFrame:

Bảng 2. So sánh tính năng và mức độ trừu tượng của RDD API và DataFrame API: (Do Kuriko IWAI tạo)

4. Tổng kết

Mục tiêu của việc tinh chỉnh Spark không phải là tối đa hóa riêng số lượng phân vùng hoặc mức sử dụng CPU.

Mục tiêu là giảm chi phí và độ trễ thực thi từ đầu đến cuối mà không gây mất ổn định.

Trong bài viết này, chúng ta tìm hiểu những mối quan hệ cốt lõi trong hệ thống:

• Tài nguyên: Worker lưu trữ executor; executor chạy các tác vụ; cấu hình CPU và bộ nhớ giới hạn mức độ chạy đồng thời.
• Lập trình bằng RDD: Các phép biến đổi tạo ra định nghĩa RDD bất biến và thông tin về dòng dõi; các phép toán hành động kích hoạt quá trình tính toán.
• Lập trình bằng DataFrame: Các phép biến đổi tạo ra những thao tác truy vấn logic; quá trình tối ưu hóa chọn các toán tử vật lý có thể thực thi; các phép toán hành động kích hoạt quá trình tính toán.
• Thực thi: Spark tổ chức công việc thành các job, stage và task; các task xử lý những phân vùng trên executor. Shuffle và các phần phụ thuộc khác tạo ra ranh giới điều phối giữa các stage.

Hiểu cách Spark thực thi các khối lượng công việc phân tán mới chỉ là nền tảng.

Bạn có thể tìm hiểu quy trình tính toán gồm bảy bước sử dụng RDD API.