PySpark 6 — MLlib: Machine Learning phân tán

Khi nào cần MLlib?

Ở PySpark tổng quan ta đã thấy Spark sinh ra để xử lý dữ liệu vượt bộ nhớ một máy bằng cách chia nhỏ và tính song song trên nhiều executor. MLlib — chính xác là gói pyspark.ml với API dựa trên DataFrame — mang đúng triết lý đó sang machine learning: thuật toán học máy được viết lại để chạy phân tán, mỗi executor xử lý một phần dữ liệu, driver tổng hợp.

Câu hỏi thực chiến đầu tiên không phải "MLlib làm được gì" mà là "khi nào tôi nên dùng nó thay vì scikit-learn". Câu trả lời gọn:

Điểm mấu chốt: nếu dữ liệu vừa một máy, gần như luôn nên .toPandas() rồi dùng scikit-learn — nó có nhiều thuật toán hơn, cộng đồng lớn hơn, gỡ lỗi dễ hơn. MLlib chỉ thắng khi quy mô là ràng buộc thật: khi việc lấy mẫu (sampling) làm mất tín hiệu quan trọng, hoặc khi bản thân feature engineering đã phải chạy trên cluster nên kéo dữ liệu về một máy là bất khả thi.

Những đánh đổi phải chấp nhận

MLlib không miễn phí. So với scikit-learn:

• Ít thuật toán hơn. MLlib có hồi quy tuyến tính/logistic, cây quyết định, Random Forest, GBT (Gradient-Boosted Trees), Naive Bayes, KMeans, ALS (gợi ý), một số clustering — nhưng không có nhiều biến thể tinh vi. Không có XGBoost bản gốc (phải dùng gói ngoài xgboost4j / SynapseML).
• Overhead phân tán. Với dữ liệu nhỏ, chi phí điều phối cluster khiến MLlib chậm hơn scikit-learn chạy trên một máy.
• Gỡ lỗi khó hơn. Lỗi phân tán, lazy evaluation, lịch trình DAG khiến việc soi model phức tạp hơn.
• Diễn giải mô hình (interpretability) hạn chế hơn. Ít công cụ SHAP/partial-dependence tích hợp sẵn.

Nói cách khác: chọn MLlib vì buộc phải phân tán, không phải vì nó "xịn hơn".

Ba khối cơ bản: Transformer, Estimator, Pipeline

Toàn bộ pyspark.ml xoay quanh ba khái niệm. Nắm chắc ba khối này là hiểu 80% MLlib.

Transformer — một đối tượng biến đổi DataFrame thành DataFrame khác qua phương thức .transform(). Nó không học gì, chỉ áp một phép biến đổi đã định. Ví dụ:

• VectorAssembler — gom nhiều cột số thành một cột vector tên features (MLlib yêu cầu đặc trưng phải nằm trong một cột vector duy nhất).
• Một Model đã huấn luyện cũng là Transformer: model.transform(df) sinh cột prediction.

Estimator — một đối tượng có phương thức .fit(), đọc dữ liệu và học ra một Model (Model đó chính là một Transformer). Ví dụ: LogisticRegression, RandomForestClassifier, GBTClassifier, KMeans, và cả một số bước tiền xử lý cần "học" tham số như StringIndexer (học bảng ánh xạ nhãn→số) hay StandardScaler (học trung bình/độ lệch chuẩn để chuẩn hóa).

Estimator.fit(df)  ─►  Model  (là Transformer)
Model.transform(df)  ─►  DataFrame có cột prediction

Pipeline — ghép nhiều stage (mỗi stage là Transformer hoặc Estimator) thành một chuỗi. Bản thân Pipeline là một Estimator: khi .fit(), nó chạy tuần tự các stage, fit những stage là Estimator và transform truyền dữ liệu qua từng bước. Kết quả là một PipelineModel — đóng gói toàn bộ chuỗi biến đổi + mô hình cuối.

Vì sao Pipeline quan trọng đến vậy?

• Chống rò rỉ dữ liệu (data leakage). StandardScaler, StringIndexer chỉ được fit trên tập train. Nếu tự tay biến đổi rời rạc, rất dễ vô tình tính trung bình/scale trên cả tập test. Pipeline đảm bảo mọi stage học trên train, áp máy móc lên test.
• Tái lập (reproducibility). Lưu một PipelineModel là lưu cả tiền xử lý lẫn mô hình. Lúc suy luận chỉ cần nạp lại và transform — không có nguy cơ quên một bước biến đổi.
• Nhất quán train/serve. Cùng một object dùng cho cả huấn luyện và production, loại bỏ lệch train–serve (train-serve skew).

Feature engineering trong MLlib

Bước tốn công nhất luôn là chuẩn bị đặc trưng. Phần lớn công việc này thực chất là DataFrame API thuần túy (lọc, join, tạo cột dẫn xuất bằng F.*), chỉ vài bước cuối mới dùng transformer của MLlib. Nguyên tắc: làm sạch và tạo đặc trưng bằng hàm built-in F.*, tránh Python UDF — vì UDF phá tối ưu Catalyst và chậm trên dữ liệu lớn, đúng cảnh mà ta chọn MLlib để xử lý.

Xử lý biến phân loại (categorical). MLlib chỉ ăn số. Chuỗi như city, product_type phải qua hai bước:

• StringIndexer: ánh xạ mỗi giá trị chuỗi thành một chỉ số nguyên ("HN"→0, "HCM"→1...). Đặt handleInvalid="keep" để nhãn lạ ở tập test không làm hỏng job.
• OneHotEncoder: biến chỉ số đó thành vector nhị phân (one-hot), tránh việc mô hình hiểu nhầm thứ tự số có ý nghĩa lớn/nhỏ. Với cây (Random Forest/GBT) đôi khi có thể bỏ one-hot vì cây xử lý được biến rời rạc, nhưng với mô hình tuyến tính thì gần như bắt buộc.

Gom đặc trưng. VectorAssembler(inputCols=[...], outputCol="features") gom tất cả cột số + vector one-hot thành cột features duy nhất. Đây là "hợp đồng" bắt buộc: mọi Estimator đọc cột features (vector) và label (nhãn số).

Chuẩn hóa (scaling). Mô hình tuyến tính, hồi quy logistic nhạy với thang đo. StandardScaler (trừ trung bình, chia độ lệch chuẩn) hoặc MinMaxScaler đưa các đặc trưng về cùng thang. Cây quyết định không cần scaling nên có thể bỏ.

Xử lý mất cân bằng nhãn (class imbalance). Trong ngân hàng, tỉ lệ khách vỡ nợ / gian lận thường rất nhỏ (1–3%). Nếu để nguyên, mô hình dễ "đoán tất cả là tốt" mà vẫn đạt accuracy 98% — vô dụng. Các cách trong MLlib:

• Trọng số mẫu (class weight). Tạo một cột trọng số (dòng thuộc lớp hiếm nhận trọng số lớn hơn) bằng F.when(...), rồi truyền vào weightCol của LogisticRegression/GBTClassifier. Đây là cách gọn và được khuyến nghị nhất vì không đổi kích thước dữ liệu.
• Lấy mẫu lại (resampling). df.sampleBy("label", fractions={...}) để down-sample lớp đa số hoặc up-sample lớp hiếm.
• Chọn đúng chỉ số đánh giá. Đừng dùng accuracy; dùng AUC, precision/recall, F1 — sẽ nói ở dưới.

Huấn luyện và đánh giá

Quy trình chuẩn: chia dữ liệu, fit Pipeline, đánh giá, tinh chỉnh siêu tham số.

Chia train/test. df.randomSplit([0.8, 0.2], seed=42). Với dữ liệu chuỗi thời gian (giao dịch theo thời gian), nên chia theo thời gian thay vì ngẫu nhiên để tránh dùng tương lai đoán quá khứ.

Đánh giá (Evaluator). MLlib cung cấp các evaluator sẵn:

• BinaryClassificationEvaluator — mặc định trả AUC (areaUnderROC), chỉ số vàng cho bài toán nhị phân mất cân bằng vì không phụ thuộc ngưỡng.
• MulticlassClassificationEvaluator — accuracy, f1, weightedPrecision/Recall cho phân loại nhiều lớp.
• RegressionEvaluator — RMSE, MAE, R² cho hồi quy.

Tinh chỉnh siêu tham số (hyperparameter tuning). Đây là chỗ MLlib rất mạnh vì tận dụng cluster để thử nhiều cấu hình song song:

• ParamGridBuilder — khai báo lưới tham số cần thử (ví dụ maxDepth ∈ {3,5,7}, maxIter ∈ {20,50}).
• CrossValidator — k-fold cross-validation: chia train thành k phần, huấn luyện k lần cho mỗi cấu hình, chọn cấu hình có điểm evaluator trung bình tốt nhất. Chính xác nhưng tốn (k × số cấu hình lần fit).
• TrainValidationSplit — chỉ chia train/validation một lần (ví dụ 75/25), rẻ hơn CrossValidator nhiều lần, phù hợp khi dữ liệu đã lớn nên một lần chia cũng đủ tin cậy.

Cả CrossValidator và TrainValidationSplit đều là Estimator: .fit() ra một model đã chọn cấu hình tốt nhất, truy cập qua .bestModel.

Ví dụ: Pipeline phân loại + CrossValidator + AUC (minh hoạ)

Đoạn code dưới là minh hoạ Python (không phải SQL sandbox), dựng một pipeline chấm điểm nhị phân đầy đủ:

from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
from pyspark.ml.classification import GBTClassifier
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
import pyspark.sql.functions as F

# df: đã join & làm sạch bằng DataFrame API. Nhãn 'default' 0/1.
df = df.withColumn("label", F.col("default").cast("double"))

# 1) Xử lý biến phân loại
idx = StringIndexer(inputCol="product_type", outputCol="product_idx",
                    handleInvalid="keep")
ohe = OneHotEncoder(inputCols=["product_idx"], outputCols=["product_vec"])

# 2) Gom đặc trưng số + one-hot thành cột 'features'
assembler = VectorAssembler(
    inputCols=["age", "income", "debt_ratio", "num_loans", "product_vec"],
    outputCol="features")

# 3) Xử lý mất cân bằng bằng trọng số mẫu
df = df.withColumn("w", F.when(F.col("label") == 1, 8.0).otherwise(1.0))

# 4) Estimator cuối
gbt = GBTClassifier(featuresCol="features", labelCol="label",
                    weightCol="w", maxIter=50)

pipe = Pipeline(stages=[idx, ohe, assembler, gbt])

# 5) Chia dữ liệu
train, test = df.randomSplit([0.8, 0.2], seed=42)

# 6) Lưới tham số + cross-validation
grid = (ParamGridBuilder()
        .addGrid(gbt.maxDepth, [3, 5, 7])
        .addGrid(gbt.maxIter, [30, 50])
        .build())

evaluator = BinaryClassificationEvaluator(
    labelCol="label", metricName="areaUnderROC")

cv = CrossValidator(estimator=pipe, estimatorParamMaps=grid,
                    evaluator=evaluator, numFolds=3, parallelism=4)

cv_model = cv.fit(train)                 # huấn luyện phân tán trên cluster

# 7) Đánh giá trên tập test độc lập
pred = cv_model.transform(test)
auc = evaluator.evaluate(pred)
print(f"AUC test = {auc:.3f}")

# 8) Lưu toàn bộ pipeline đã chọn cấu hình tốt nhất
cv_model.bestModel.write().overwrite().save("s3://ncb-ml/risk/pipeline_v3")

Chú ý: parallelism=4 cho CrossValidator huấn luyện nhiều fold/cấu hình song song. weightCol="w" xử lý mất cân bằng ngay trong estimator. Toàn bộ cv.fit(train) chạy phân tán — dữ liệu không bao giờ bị kéo về driver.

Lưu, nạp và suy luận quy mô lớn

Một PipelineModel là tài sản có thể tuần tự hóa. Lưu bằng .save(path), nạp lại bằng PipelineModel.load(path):

from pyspark.ml import PipelineModel

model = PipelineModel.load("s3://ncb-ml/risk/pipeline_v3")

# Suy luận trên toàn bộ khách hàng mới - hàng chục triệu dòng
scored = model.transform(new_customers_df)
scored.select("customer_id", "prediction", "probability").write \
      .mode("overwrite").parquet("s3://ncb-ml/risk/scores_2026_07")

Điểm mạnh: suy luận cũng phân tán. model.transform() trên tập cực lớn được chia cho toàn cluster, và vì đã đóng gói cả tiền xử lý, dữ liệu mới chỉ cần đúng các cột đầu vào thô — mọi bước index/scale/assemble tự chạy lại y hệt lúc train. Cột probability (vector xác suất mỗi lớp) thường hữu ích hơn prediction (nhãn 0/1) vì cho phép nghiệp vụ tự chọn ngưỡng theo khẩu vị rủi ro.

Đưa ra production (MLOps). Vài lưu ý ngắn:

• Batch scoring là mô hình triển khai tự nhiên nhất của MLlib: một job Spark theo lịch (Airflow) nạp model, transform dữ liệu ngày, ghi kết quả ra kho.
• Phiên bản hóa (versioning). Đặt tên đường dẫn có version (pipeline_v3) hoặc dùng MLflow để đăng ký model, ghi lại tham số/metric.
• Serving độ trễ thấp (real-time API từng khách) không phải thế mạnh MLlib — Spark khởi động nặng. Khi cần, thường export logic sang định dạng nhẹ hơn (ONNX/PMML qua thư viện ngoài) hoặc huấn luyện lại bằng framework serving-friendly. MLlib mạnh nhất ở học trên dữ liệu khổng lồ và scoring theo lô.

Use case thực tế

Bài toán NCB — mô hình chấm điểm rủi ro tín dụng (PD) trên toàn danh mục.

Khối Quản trị rủi ro cần một mô hình xác suất vỡ nợ (Probability of Default) cho toàn bộ khách hàng vay, không chỉ mẫu. Dữ liệu huấn luyện gộp từ nhiều nguồn: lịch sử khoản vay, giao dịch tài khoản, thông tin nhân khẩu — tổng cộng khoảng 40 triệu dòng khách × lịch sử sau khi làm phẳng, cỡ vài trăm GB. Kéo về một máy để chạy scikit-learn là bất khả thi, và lấy mẫu 1% làm mất tín hiệu ở nhóm khách hiếm gặp (vay lớn, ngành đặc thù) — đúng nhóm rủi ro cao nhất. Đây là ca dùng MLlib điển hình.

Cách làm:

1. Feature engineering trên Spark. Dùng DataFrame API join khoản vay với đặc trưng hành vi giao dịch (số dư trung bình 6 tháng, số lần trễ hạn, tỉ lệ nợ/thu nhập). Toàn bộ bằng F.*, không UDF.
2. Pipeline. StringIndexer + OneHotEncoder cho ngành nghề/sản phẩm; VectorAssembler gom ~35 đặc trưng; GBTClassifier với weightCol xử lý tỉ lệ vỡ nợ chỉ ~2.5%.
3. Tuning. CrossValidator 3-fold trên lưới maxDepth × maxIter, evaluator theo AUC. Chạy trên cluster ~20 executor, hoàn tất trong khoảng 1–2 giờ (con số ước lượng, tùy cấu hình cluster).
4. Đánh giá. AUC trên tập test độc lập quanh mức 0.80–0.83 (con số minh hoạ/ước lượng, không phải kết quả đo thật) — đủ để phân tầng rủi ro. So sánh: mô hình cũ chạy trên mẫu 5% có AUC thấp hơn và bất ổn giữa các lần huấn luyện.
5. Triển khai. Lưu PipelineModel lên S3, một DAG Airflow hằng tháng nạp model, transform toàn bộ khách, ghi cột probability ra kho để đội tín dụng dùng làm đầu vào quyết định hạn mức.

Giá trị cốt lõi: học trên toàn bộ dữ liệu, không lấy mẫu, nên nắm được cả những nhóm khách hiếm mà mẫu ngẫu nhiên bỏ sót. Cùng logic áp dụng cho mô hình dự đoán rời bỏ (churn) — nhãn là "đóng tài khoản trong 90 ngày tới", đặc trưng là xu hướng số dư và tần suất giao dịch giảm dần.

Ghi nhớ

• Chọn MLlib vì buộc phải phân tán, không phải vì nó xịn hơn. Dữ liệu vừa một máy → scikit-learn (nhiều thuật toán, dễ gỡ lỗi hơn). Dữ liệu quá lớn / không được lấy mẫu → MLlib.
• Ba khối: Transformer (.transform, không học), Estimator (.fit ra Model), Pipeline (ghép stage). Model là một Transformer.
• Pipeline chống rò rỉ dữ liệu và đảm bảo tái lập — mọi bước tiền xử lý học trên train, áp máy móc lên test và lúc serving; đóng gói cùng mô hình.
• MLlib chỉ ăn số: StringIndexer → OneHotEncoder cho biến phân loại; VectorAssembler gom thành một cột features; StandardScaler cho mô hình tuyến tính.
• Mất cân bằng nhãn (vỡ nợ/gian lận hiếm): ưu tiên weightCol; đánh giá bằng AUC / precision-recall / F1, tuyệt đối không dùng accuracy.
• Tuning: ParamGridBuilder + CrossValidator (k-fold, chính xác, tốn) hoặc TrainValidationSplit (một lần chia, rẻ). Lấy .bestModel.
• Làm sạch/tạo đặc trưng bằng F.* built-in, tránh Python UDF vì nó phá tối ưu trên dữ liệu lớn.
• Lưu/nạp PipelineModel để suy luận phân tán trên tập cực lớn; MLlib mạnh ở batch scoring, không hợp serving độ trễ thấp — khi cần thì export/MLflow.

Nguồn tham khảo

• Apache Spark Documentation — MLlib: Main Guide (Pipelines, Transformers, Estimators): spark.apache.org/docs/latest/ml-guide.html
• Apache Spark Documentation — Extracting, transforming and selecting features (StringIndexer, OneHotEncoder, VectorAssembler, StandardScaler): spark.apache.org/docs/latest/ml-features.html
• Apache Spark Documentation — Classification and Regression (LogisticRegression, GBTClassifier, RandomForest): spark.apache.org/docs/latest/ml-classification-regression.html
• Apache Spark Documentation — ML Tuning: model selection and hyperparameter tuning (CrossValidator, TrainValidationSplit, ParamGridBuilder): spark.apache.org/docs/latest/ml-tuning.html
• PySpark API Reference — pyspark.ml package: spark.apache.org/docs/latest/api/python/reference/pyspark.ml.html
• Jules S. Damji, Brooke Wenig, Tathagata Das, Denny Lee — "Learning Spark: Lightning-Fast Data Analytics", 2nd Edition, O'Reilly (2020), Chapter 10: Machine Learning with MLlib
• MLflow Documentation — Model Registry: mlflow.org/docs/latest/model-registry.html