Bài 14: Sklearn Pipeline — đóng gói chuỗi preprocessing

Sau bài này, bạn sẽ:

• Hiểu vì sao tách rời preprocessing và model dễ gây leak, sai lệch giữa train và predict.
• Dùng được Pipeline, make_pipeline, ColumnTransformer, make_column_selector.
• Truy cập step bên trong pipeline qua named_steps và indexing.
• Lồng ColumnTransformer vào Pipeline để gom toàn bộ workflow vào 1 object.
• Save / load pipeline với joblib để dùng lại ở serving.
• Biết cách đặt tham số grid cho từng step qua cú pháp step__param.

Một workflow ML thực tế gồm nhiều bước nối tiếp: impute missing → encode categorical → scale numerical → (đôi khi) feature selection / PCA → model. Code "rời" thường có dạng:

imputer.fit(X_train)
X_train = imputer.transform(X_train)

encoder.fit(X_train)
X_train = encoder.transform(X_train)

scaler.fit(X_train)
X_train = scaler.transform(X_train)

model.fit(X_train, y_train)

# inference / test
X_test = imputer.transform(X_test)
X_test = encoder.transform(X_test)
X_test = scaler.transform(X_test)
y_pred = model.predict(X_test)

Bốn vấn đề thường gặp:

• Quên áp 1 bước lên test: forget scaler.transform(X_test) → model nhận distribution lệch → metric tụt mà không hiện rõ nguyên nhân.
• Data leak: lỡ tay scaler.fit(X) trên toàn bộ data (cả test) → mean/std của test rỉ vào train. Bài 7 và 11 đã nhắc.
• Khó cross-validation đúng: mỗi fold cần fit lại preprocessor trên train fold mới, không phải fit 1 lần rồi reuse.
• Khó reuse train/predict: code train và code serving phải đồng bộ thứ tự bước; lệch 1 dòng là production sai khác lab.

Pipeline đóng gói toàn bộ chuỗi thành một estimator duy nhất. Chỉ cần pipe.fit(X_train, y_train) và pipe.predict(X_test) — sklearn tự gọi đúng fit/transform đúng thứ tự.

API ở sklearn.pipeline.Pipeline nhận list of tuples (name, estimator):

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression

pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("model", LogisticRegression()),
])

pipe.fit(X_train, y_train)
y_pred = pipe.predict(X_test)
score = pipe.score(X_test, y_test)

Quy tắc về cấu trúc step:

• Tất cả step trừ step cuối phải là transformer — có fit và transform. Vd: StandardScaler, OneHotEncoder, SimpleImputer, PCA.
• Step cuối có thể là transformer (pipeline-as-transformer) hoặc estimator có predict (model). Khi step cuối là model, pipeline đóng vai trò estimator end-to-end.
• Tên mỗi step (chuỗi) phải unique và không chứa hai dấu gạch dưới liên tiếp __ (vì __ dùng cho grid param).

Khi gọi pipe.fit(X_train, y_train), sklearn lần lượt:

1. scaler.fit_transform(X_train) → X_scaled.
2. model.fit(X_scaled, y_train).

Khi gọi pipe.predict(X_test):

1. scaler.transform(X_test) (không fit lại) → X_scaled.
2. model.predict(X_scaled).

Đây chính là cách Pipeline ngăn data leak: mọi bước fit chỉ xảy ra trên data truyền vào pipe.fit.

Nếu lười đặt tên, dùng make_pipeline — tự đặt tên step theo class name lowercase:

from sklearn.pipeline import make_pipeline

pipe = make_pipeline(StandardScaler(), LogisticRegression())
print(pipe.steps)
# [('standardscaler', StandardScaler()), ('logisticregression', LogisticRegression())]

Khi nào dùng cái nào:

• Pipeline([...]): production code, hoặc khi cần đặt tên dễ đọc để GridSearch ("scaler__with_mean" trực quan hơn "standardscaler__with_mean").
• make_pipeline(...): notebook, prototype nhanh, ít hyperparameter cần tune.

Nếu trong cùng pipeline có 2 instance cùng class (vd 2 StandardScaler ở 2 vị trí), make_pipeline tự thêm hậu tố -1, -2; lúc đó nên chuyển sang Pipeline([...]) để rõ ràng.

Sau khi fit, có thể inspect từng step:

pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("model", LogisticRegression()),
])
pipe.fit(X_train, y_train)

# Cách 1: named_steps (dict-like)
print(pipe.named_steps["scaler"].mean_)
print(pipe.named_steps["model"].coef_)

# Cách 2: indexing trực tiếp
print(pipe["scaler"].scale_)
print(pipe["model"].classes_)

# Cách 3: slicing — lấy sub-pipeline
preproc_only = pipe[:-1]   # bỏ step cuối, còn lại transformer
print(preproc_only.transform(X_test).shape)

Slicing đặc biệt hữu dụng khi muốn export chỉ phần preprocessing (vd cho inference service tách biệt) hoặc kiểm tra output sau preprocessing trước khi vào model.

Dataset thực tế gồm cả cột số và cột categorical, mỗi loại cần transformer khác nhau. ColumnTransformer là thành phần quan trọng nhất khi build pipeline cho tabular data:

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder

numerical_cols = ["age", "fare"]
categorical_cols = ["sex", "embarked"]

preprocessor = ColumnTransformer(
    transformers=[
        ("num", StandardScaler(), numerical_cols),
        ("cat", OneHotEncoder(handle_unknown="ignore"), categorical_cols),
    ],
    remainder="drop",   # cột không khai báo: drop / passthrough / transformer
)

X_transformed = preprocessor.fit_transform(df)
print(preprocessor.get_feature_names_out())

Mỗi tuple có dạng (name, transformer, columns). columns có thể là:

• List tên cột (DataFrame): ["age", "fare"].
• List index (numpy array): [0, 3, 5].
• Slice: slice(0, 4).
• Callable / selector: dùng với make_column_selector ở mục 7.

Tham số remainder:

• "drop" (mặc định) — bỏ cột không khai báo. An toàn nhưng dễ bất ngờ nếu quên 1 cột.
• "passthrough" — giữ nguyên cột không khai báo, nối thêm vào output.
• Một transformer (vd StandardScaler()) — áp lên tất cả cột còn lại.

Output của fit_transform là numpy.ndarray hoặc scipy.sparse tuỳ transformer (vd OneHotEncoder(sparse_output=True) → sparse). Thứ tự cột output theo thứ tự khai báo trong transformers, không theo thứ tự cột gốc.

make_column_transformer bỏ qua phần đặt tên — hoạt động giống make_pipeline:

from sklearn.compose import make_column_transformer

preprocessor = make_column_transformer(
    (StandardScaler(), numerical_cols),
    (OneHotEncoder(handle_unknown="ignore"), categorical_cols),
    remainder="drop",
)

Tên step tự sinh: "standardscaler", "onehotencoder".

make_column_selector chọn cột theo dtype — hữu ích khi dataset có nhiều cột và bạn không muốn liệt kê tay:

import numpy as np
from sklearn.compose import ColumnTransformer, make_column_selector

preprocessor = ColumnTransformer([
    ("num", StandardScaler(), make_column_selector(dtype_include=np.number)),
    ("cat", OneHotEncoder(handle_unknown="ignore"),
        make_column_selector(dtype_include=object)),
])

preprocessor.fit_transform(df)

Tham số:

• dtype_include / dtype_exclude — lọc theo dtype (np.number, object, "category", bool...).
• pattern — regex tên cột, vd pattern="^feat_".

Lưu ý: selector chỉ chạy lúc fit; sau khi fit, tập cột được "khoá". Nếu X_test có cấu trúc cột khác (thiếu cột, đổi dtype), transform sẽ lỗi — đúng hành vi mong muốn để tránh schema drift im lặng.

Pattern thực dụng nhất khi build ML cho tabular: ColumnTransformer là 1 step trong Pipeline lớn cùng với model.

from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression

num_pipe = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("scaler", StandardScaler()),
])

cat_pipe = Pipeline([
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("onehot", OneHotEncoder(handle_unknown="ignore")),
])

preprocessor = ColumnTransformer([
    ("num", num_pipe, numerical_cols),
    ("cat", cat_pipe, categorical_cols),
])

full_pipe = Pipeline([
    ("preprocessor", preprocessor),
    ("model", LogisticRegression(max_iter=1000)),
])

full_pipe.fit(X_train, y_train)
print(full_pipe.score(X_test, y_test))

Hai mức Pipeline: pipeline con (impute → scale) cho từng nhóm cột, gói trong ColumnTransformer, gói tiếp trong pipeline chính cùng model. Tất cả gọi qua một entry point duy nhất.

Khi full_pipe.fit(X_train, y_train):

1. ColumnTransformer chia cột → num_pipe và cat_pipe fit trên train.
2. Output concatenate thành matrix.
3. Model fit trên matrix đó.

Khi full_pipe.predict(X_new): chỉ transform (không fit), giữ đúng mean/std và tập category đã học từ train.

• Chống data leak by design: scaler/encoder chỉ fit trong pipe.fit trên train; pipe.predict hay cross_val_score chỉ gọi transform. Không có chỗ nào "lỡ tay" fit trên toàn bộ data.
• Cross-validation chuẩn: cross_val_score(pipe, X, y, cv=5) tự fit lại preprocessor trên train fold cho mỗi fold → metric ước lượng không bị optimistic bias.
• Hyperparameter tuning: GridSearch / RandomSearch tune được tham số của cả preprocessor lẫn model trong cùng grid (xem mục 10).
• Reproducibility: 1 file .pkl chứa toàn bộ workflow. Service load file đó là gọi .predict được luôn, không cần copy lại code preprocessing.
• Đơn giản code path: train và serving cùng dùng 1 object → không bị lệch giữa "code trong notebook" và "code trong API".

Tham số của step trong pipeline được truy cập qua cú pháp step_name__param_name (double underscore __). Ví dụ tune cả scaler lẫn regularization:

from sklearn.model_selection import GridSearchCV

pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("model", LogisticRegression(max_iter=1000)),
])

param_grid = {
    "scaler__with_mean": [True, False],
    "model__C": [0.1, 1.0, 10.0],
}

search = GridSearchCV(pipe, param_grid, cv=5, scoring="accuracy")
search.fit(X_train, y_train)

print(search.best_params_)
# {'model__C': 1.0, 'scaler__with_mean': True}
print(search.best_score_)

Khi step là pipeline lồng pipeline (vd preprocessor__num__scaler__with_mean), cú pháp __ chuỗi xuyên qua các tầng. Bài 40 sẽ deep về GridSearchCV / RandomizedSearchCV và Bayesian search.

Sau khi pipeline đã fit, lưu thành file để serving:

import joblib

# Train xong, save toàn bộ pipeline
joblib.dump(full_pipe, "model.pkl")

# Ở môi trường khác (API, batch job)
pipe_loaded = joblib.load("model.pkl")
y_pred = pipe_loaded.predict(X_new)

Vì pipeline chứa cả preprocessing lẫn model, file .pkl đủ cho inference — không cần repo code preprocessing kèm theo.

Lưu ý quan trọng khi deploy:

• Version sklearn: file pickle phụ thuộc cấu trúc class. Load trên version khác (đặc biệt khác minor) có thể warning hoặc fail. Pin version giữa training và serving.
• Python version: pickle protocol khác nhau giữa Python 2/3, một số khác biệt nhỏ giữa 3.x. Khuyến nghị cùng minor version.
• Schema input: pipeline expect đúng tập cột (đúng tên, đúng dtype) như lúc train. Validate schema input trước khi gọi predict.
• Bảo mật: pickle có thể chạy code tuỳ ý khi load — không load file .pkl từ nguồn không tin cậy.

So với pickle chuẩn, joblib hiệu quả hơn cho object numpy lớn (dùng memmap, nén). Sklearn doc chính thức khuyến nghị joblib cho model.

• Quên truyền y vào pipe.fit: pipe.fit(X_train) không lỗi rõ ràng nếu step cuối nhận được None — vài transformer chấp nhận y=None. Nhưng model step sẽ fail sâu trong stack. Luôn viết pipe.fit(X_train, y_train).
• Truyền 1 cột Series 1D vào ColumnTransformer: df["age"] (Series) khác df[["age"]] (DataFrame). ColumnTransformer cần input 2D. Dùng double brackets hoặc df[["age"]].
• Đặt scaler sau OneHotEncoder trong cùng pipeline tổng: scale cột 0/1 không sai về toán nhưng làm coefficient khó diễn giải. Thường tách riêng num_pipe (scale) và cat_pipe (one-hot), nối qua ColumnTransformer.
• Fit lại pipeline trên test: pipe.fit(X_test) ghi đè trạng thái — không bao giờ làm. Chỉ predict / transform / score trên test.
• Pipeline có step nhận y: vài transformer (vd TargetEncoder) cần y để fit. Trong pipeline, sklearn tự forward y đến mọi step nên không cần làm thủ công, nhưng phải chắc step đó hỗ trợ fit(X, y) với y ≠ None.
• Lưu pipeline chứa lambda hoặc closure: FunctionTransformer(lambda x: ...) không pickle được trong nhiều môi trường. Định nghĩa hàm top-level rồi truyền vào.

Một pipeline thực tế cho bài Titanic — gom impute, scale, one-hot, model, save/load:

import pandas as pd
import numpy as np
import joblib

from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer, make_column_selector
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression

# 1. Load data (giả định đã có CSV)
df = pd.read_csv("titanic.csv")
features = ["pclass", "sex", "age", "sibsp", "parch", "fare", "embarked"]
X = df[features]
y = df["survived"]

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y,
)

# 2. Sub-pipeline cho cột số và cột categorical
num_pipe = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("scaler", StandardScaler()),
])

cat_pipe = Pipeline([
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("onehot", OneHotEncoder(handle_unknown="ignore")),
])

# 3. ColumnTransformer chọn cột theo dtype
preprocessor = ColumnTransformer([
    ("num", num_pipe, make_column_selector(dtype_include=np.number)),
    ("cat", cat_pipe, make_column_selector(dtype_include=object)),
])

# 4. Pipeline tổng = preprocessor + model
full_pipe = Pipeline([
    ("preprocessor", preprocessor),
    ("model", LogisticRegression(max_iter=1000)),
])

# 5. Cross-validation đúng cách — preprocessor fit lại mỗi fold
cv_scores = cross_val_score(full_pipe, X_train, y_train, cv=5, scoring="accuracy")
print(f"CV accuracy: {cv_scores.mean():.3f} ± {cv_scores.std():.3f}")

# 6. Fit cuối trên toàn bộ train
full_pipe.fit(X_train, y_train)
print(f"Test accuracy: {full_pipe.score(X_test, y_test):.3f}")

# 7. Inspect step
print(full_pipe.named_steps["model"].coef_.shape)

# 8. Save / load
joblib.dump(full_pipe, "titanic_pipeline.pkl")
pipe_loaded = joblib.load("titanic_pipeline.pkl")
print(pipe_loaded.predict(X_test.iloc[:5]))

Toàn bộ workflow nằm trong full_pipe. File titanic_pipeline.pkl là deliverable cho serving — không cần kèm code tiền xử lý.

Module 2 đã đi qua các nhóm chủ đề chính của bước "chuẩn bị dữ liệu":

• Split dữ liệu (Bài 6): train / validation / test — vì sao 3 tập, cách stratify, random_state.
• Scaling (Bài 7-9): Min-Max, Standardization, lựa chọn theo model và phân phối.
• Encoding categorical (Bài 10-11): One-Hot, Label/Ordinal — nominal vs ordinal, handle_unknown cho production.
• Missing value và outlier (Bài 12): impute, IQR/z-score, khi nào xoá / clip / giữ.
• Feature engineering (Bài 13): tạo feature mới từ kiến thức domain — polynomial, interaction, datetime.
• Pipeline (bài này): gom tất cả thành 1 object reproducible.

Pattern xuyên suốt: fit chỉ trên train, transform trên test. Pipeline làm việc đó tự động — đây là lý do mọi Module sau (3: model classic, 4: model nâng cao, 5: validation, 6: project) đều dùng Pipeline làm khung cơ sở.

Một số chủ đề chưa cover ở Module 2 và sẽ quay lại sau:

• Feature selection / dimensionality reduction (PCA, mutual_info, L1) — Module 4.
• Imbalanced data (SMOTE, class_weight) — Module 5 cùng metric F1/PR-AUC.
• Target Encoding nâng cao — Module 4 khi gặp dataset cardinality cao.

Module 3 bắt đầu với Linear Regression — model đầu tiên áp dụng đầy đủ những gì Module 2 chuẩn bị.

Bài 1. Trên dataset Iris (sklearn.datasets.load_iris), tạo Pipeline gồm StandardScaler → LogisticRegression. Fit trên train, in accuracy trên test. So sánh với make_pipeline — verify pipe.steps khác tên thế nào.

Bài 2. Sinh DataFrame:

df = pd.DataFrame({
    "x1": [1.0, 2.0, np.nan, 4.0, 5.0],
    "x2": [10, 20, 30, 40, 50],
    "cat": ["a", "b", "a", "c", "b"],
})
y = [0, 1, 0, 1, 1]

Viết ColumnTransformer apply SimpleImputer(strategy="median") + StandardScaler cho x1, x2; và OneHotEncoder(handle_unknown="ignore") cho cat. Fit, in get_feature_names_out().

Bài 3. Lồng ColumnTransformer ở Bài 2 vào một Pipeline tổng có LogisticRegression ở cuối. Save thành pipe.pkl bằng joblib.dump. Mở Python session mới, load lại, predict trên 1 sample bất kỳ.

Bài 4. Cho pipeline ở Bài 3, định nghĩa param_grid:

param_grid = {
    "preprocessor__num__scaler__with_mean": [True, False],
    "model__C": [0.01, 0.1, 1.0, 10.0],
}

Chạy GridSearchCV với cv=3, in best_params_ và best_score_. Trả lời: cú pháp preprocessor__num__scaler__with_mean nghĩa là gì?

Gợi ý đáp án bài 4: chuỗi __ xuyên qua các tầng — pipeline tổng có step preprocessor (ColumnTransformer), bên trong có sub-pipeline tên num, bên trong có step scaler (StandardScaler), tham số with_mean. Mỗi tầng cách nhau bằng __.

Bài 15: Linear Regression là gì — model đầu tiên của Module 3. Áp dụng pipeline pattern vừa học vào bài toán regression: từ giả định tuyến tính, hàm loss MSE, cho đến nghiệm closed-form và gradient descent.