scikit-learn Pipeline ใน Python: คู่มือสร้าง ML Pipeline ฉบับสมบูรณ์ (2026)

อัปเดต: 7 มิถุนายน 2026

scikit-learn Pipeline คือเครื่องมือที่ใช้รวมขั้นตอนการประมวลผลข้อมูล (preprocessing) และการเทรนโมเดล Machine Learning ให้อยู่ในวัตถุเดียวกัน ทำให้โค้ดสะอาด ลดข้อผิดพลาด และที่สำคัญคือป้องกัน data leakage ระหว่างทำ cross-validation บทความนี้จะพาคุณสร้าง ML Pipeline ตั้งแต่พื้นฐานจนถึงการใช้งาน ColumnTransformer, FeatureUnion, และ GridSearchCV ในเวอร์ชัน scikit-learn 1.8 (ปี 2026) พร้อมโค้ดที่รันได้จริง

ผมเองเคยเจอบั๊กตัวคลาสสิกตอนส่งโมเดล Titanic ขึ้น staging ครั้งแรก คือ fit_transform ไปทั้ง dataset ก่อนแบ่ง train/test ครับ คะแนน ROC AUC ดูสวยงามมากตอน dev แต่พอเจอข้อมูลจริงตัวเลขร่วงทันที ตั้งแต่วันนั้นผมไม่เขียน ML code โดยไม่ห่อด้วย Pipeline อีกเลย

• scikit-learn Pipeline ช่วยให้ขั้นตอน preprocessing และ model training ทำงานต่อเนื่องในวัตถุเดียว เรียก .fit() และ .predict() ได้เหมือนโมเดลปกติ
• การใช้ Pipeline ร่วมกับ cross_val_score ป้องกัน data leakage จากการ scale หรือ impute ข้อมูล validation
• ColumnTransformer ใช้สำหรับประมวลผลคอลัมน์ตัวเลขและคอลัมน์ categorical ด้วยขั้นตอนต่างกันภายใน pipeline เดียว
• scikit-learn 1.8 (มกราคม 2026) รองรับ set_output(transform="pandas") ทำให้ผลลัพธ์ของแต่ละ step เป็น DataFrame ที่อ่านง่าย
• ใช้ GridSearchCV กับ Pipeline เพื่อหา hyperparameter ที่เหมาะสมทั้งของ transformer และ estimator พร้อมกัน
• บันทึก Pipeline ทั้งก้อนด้วย joblib.dump() เพื่อนำไป deploy ใน production ได้ทันที

scikit-learn Pipeline คืออะไรและทำไมต้องใช้

scikit-learn Pipeline เป็นคลาสที่อยู่ใน sklearn.pipeline ทำหน้าที่ร้อยขั้นตอนการประมวลผลข้อมูล (transformer) หลายชั้นเข้าด้วยกัน จนถึง estimator ตัวสุดท้าย เมื่อเรียก fit() ระบบจะรัน fit_transform() ของทุก step ตามลำดับ และเก็บสถานะที่เรียนรู้ไว้ภายในวัตถุเดียว

ก่อนจะมี Pipeline นักพัฒนามักเขียนโค้ดประมาณนี้:

scaler = StandardScaler()
X_train_s = scaler.fit_transform(X_train)
X_test_s  = scaler.transform(X_test)

imputer = SimpleImputer()
X_train_i = imputer.fit_transform(X_train_s)
X_test_i  = imputer.transform(X_test_s)

model = LogisticRegression().fit(X_train_i, y_train)
y_pred = model.predict(X_test_i)

โค้ดแบบนี้มีปัญหา 3 ข้อใหญ่ คือ (1) ต้องจำลำดับการ transform() ให้ตรงกันทั้ง train และ test (2) เสี่ยงต่อ data leakage ถ้านำ fit_transform ไปใช้กับ test set โดยพลั้งเผลอ และ (3) ยุ่งยากเมื่อต้องการทำ cross-validation เพราะต้อง fit ใหม่ทุก fold

Pipeline แก้ปัญหานี้โดยห่อทั้งหมดเป็นวัตถุเดียวที่มี interface เหมือน estimator ปกติ ทำให้เข้ากันได้กับ cross_val_score, GridSearchCV, และ joblib ได้ทันที

สร้าง Pipeline พื้นฐานด้วย make_pipeline

วิธีสร้าง Pipeline ที่เร็วที่สุดคือใช้ make_pipeline() ซึ่งจะตั้งชื่อ step ให้อัตโนมัติจากชื่อคลาส (เป็น lowercase) ตัวอย่างด้านล่างคือ pipeline ทำนายความน่าจะเป็นการรอดชีวิตของผู้โดยสาร Titanic ด้วยข้อมูลตัวเลขล้วน:

import pandas as pd
from sklearn.datasets import fetch_openml
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression

titanic = fetch_openml("titanic", version=1, as_frame=True)
X = titanic.data[["age", "fare", "sibsp", "parch"]]
y = (titanic.target == "1").astype(int)

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

pipe = make_pipeline(
    SimpleImputer(strategy="median"),
    StandardScaler(),
    LogisticRegression(max_iter=1000),
)

pipe.fit(X_train, y_train)
print("Accuracy:", pipe.score(X_test, y_test))

หากต้องการตั้งชื่อ step เอง (เช่นจะอ้างถึงใน GridSearchCV) ให้ใช้ Pipeline([...]) โดยตรง:

from sklearn.pipeline import Pipeline

pipe = Pipeline([
    ("impute", SimpleImputer(strategy="median")),
    ("scale",  StandardScaler()),
    ("clf",    LogisticRegression(max_iter=1000)),
])

การเข้าถึง step ภายในทำได้ผ่าน pipe.named_steps["clf"] หรือดัชนี pipe[-1] ก็ได้ และทุกอย่างที่ pipeline เรียนรู้ไว้ (เช่น mean ของ StandardScaler) จะถูกเก็บอยู่ภายใน step นั้นๆ ไม่กระจัดกระจาย

ColumnTransformer สำหรับข้อมูลผสม

โลกจริงมักมีทั้งคอลัมน์ตัวเลขและคอลัมน์ categorical อยู่ในตารางเดียวกัน ColumnTransformer คือคำตอบสำหรับสถานการณ์นี้ครับ มันใช้ pipeline ย่อยกับแต่ละกลุ่มคอลัมน์ที่เรากำหนด แล้วนำผลลัพธ์มารวมกันโดยอัตโนมัติ ดูตัวอย่างใช้กับ Titanic แบบเต็ม:

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

numeric_cols = ["age", "fare", "sibsp", "parch"]
categorical_cols = ["sex", "embarked", "pclass"]

numeric_pipe = Pipeline([
    ("impute", SimpleImputer(strategy="median")),
    ("scale",  StandardScaler()),
])

categorical_pipe = Pipeline([
    ("impute", SimpleImputer(strategy="most_frequent")),
    ("onehot", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
])

preprocessor = ColumnTransformer([
    ("num", numeric_pipe,     numeric_cols),
    ("cat", categorical_pipe, categorical_cols),
])

clf = Pipeline([
    ("prep", preprocessor),
    ("model", LogisticRegression(max_iter=1000)),
]).set_output(transform="pandas")

clf.fit(X_train, y_train)
print("Accuracy:", clf.score(X_test, y_test))

ข้อดีของ ColumnTransformer คือเราใช้ handle_unknown="ignore" ที่ OneHotEncoder เพื่อจัดการ category ที่ไม่เคยเห็นใน training (เช่นมีพอร์ตขึ้นเรือใหม่ใน test set) โดยไม่ต้องเขียน try/except เอง การจัดการข้อมูลที่หายไปก็ทำในระดับ pipeline ย่อย หากต้องการเทคนิคจัดการ missing data ที่ลึกขึ้น โปรดดู คู่มือจัดการข้อมูลหายไปใน pandas ที่ครอบคลุม dropna, fillna, และ interpolate

ป้องกัน Data Leakage ด้วย Pipeline และ Cross-Validation

Data leakage คือสถานการณ์ที่ข้อมูลจาก validation set แอบรั่วเข้ามาในขั้นตอนเรียนรู้ ตัวอย่างที่พบบ่อยที่สุดคือการ fit_transform ของ StandardScaler หรือ SimpleImputer บนข้อมูลทั้งก้อนก่อนแบ่ง train/test ทำให้สถิติ (mean, median) ของ test set ปนเข้ามาในขั้น scaling

เมื่อใช้ Pipeline ร่วมกับ cross_val_score ระบบจะ fit transformer แค่บน training fold ของแต่ละ split อัตโนมัติ:

from sklearn.model_selection import cross_val_score

scores = cross_val_score(clf, X, y, cv=5, scoring="roc_auc")
print(f"ROC AUC: {scores.mean():.3f} ± {scores.std():.3f}")

เปรียบเทียบกับการ scale ก่อน split (ผิด):

# BAD: scale ทั้งก้อนก่อนแบ่ง → leakage
X_scaled = StandardScaler().fit_transform(X[numeric_cols])
# คะแนนจะดูดีเกินจริงเพราะโมเดลเห็น mean ของ test set แล้ว

ปรับ Hyperparameter ด้วย GridSearchCV บน Pipeline

หนึ่งในความสามารถที่ทรงพลังที่สุดของ Pipeline คือเราสามารถใช้ GridSearchCV ปรับ hyperparameter ของทั้ง preprocessing และ model พร้อมกัน โดยใช้ syntax <step_name>__<param> ดับเบิลอันเดอร์สกอร์:

from sklearn.model_selection import GridSearchCV
from sklearn.ensemble import RandomForestClassifier

clf = Pipeline([
    ("prep", preprocessor),
    ("model", RandomForestClassifier(random_state=42)),
])

param_grid = {
    "prep__num__impute__strategy": ["mean", "median"],
    "model__n_estimators": [100, 300, 500],
    "model__max_depth": [None, 5, 10],
}

grid = GridSearchCV(clf, param_grid, cv=5, scoring="roc_auc", n_jobs=-1)
grid.fit(X_train, y_train)
print("Best params:", grid.best_params_)
print("Best ROC AUC:", grid.best_score_)

สังเกตว่า prep__num__impute__strategy หมายถึง: step ชื่อ prep (ซึ่งเป็น ColumnTransformer) → sub-pipeline ชื่อ num → step ชื่อ impute → parameter ชื่อ strategy ความสามารถนี้ทำให้เราเปรียบเทียบ preprocessing strategy ได้อย่างยุติธรรมในรอบ CV เดียวกัน หากต้องการความเร็วสูงขึ้น ใช้ HalvingGridSearchCV หรือ HalvingRandomSearchCV ที่เปิดใช้งานได้ตั้งแต่ scikit-learn 1.0 และถ้า DataFrame ของคุณช้าจน CV รอบหนึ่งกินเวลานานเกินไป ลองดู 10 เทคนิคเร่งความเร็ว pandas DataFrame ก่อนเข้าสู่ขั้นเทรนนะครับ

สร้าง Custom Transformer ของตัวเอง

เมื่อ transformer สำเร็จรูปไม่ตอบโจทย์ คุณสามารถสร้าง transformer เองโดย subclass BaseEstimator และ TransformerMixin ต้องมี method fit() ที่คืน self และ transform() ที่คืน array หรือ DataFrame:

from sklearn.base import BaseEstimator, TransformerMixin
import numpy as np

class LogFareTransformer(BaseEstimator, TransformerMixin):
    """ใช้ log1p กับคอลัมน์ fare เพื่อลด skewness"""
    def fit(self, X, y=None):
        return self

    def transform(self, X):
        X = X.copy()
        X["fare"] = np.log1p(X["fare"].fillna(0))
        return X

    def get_feature_names_out(self, input_features=None):
        return np.asarray(input_features)

clf = Pipeline([
    ("log_fare", LogFareTransformer()),
    ("prep",     preprocessor),
    ("model",    LogisticRegression(max_iter=1000)),
])

สำหรับ transformation แบบฟังก์ชันล้วน (stateless) ใช้ FunctionTransformer จะเขียนน้อยกว่ามาก:

from sklearn.preprocessing import FunctionTransformer

log_transform = FunctionTransformer(np.log1p, feature_names_out="one-to-one")

บันทึกและ Deploy Pipeline สู่ Production

เมื่อเทรนเสร็จ คุณบันทึกทั้ง Pipeline (รวม preprocessing + model) ลงไฟล์เดียวด้วย joblib ได้:

import joblib

joblib.dump(clf, "titanic_pipeline.joblib", compress=3)

# ใน production
model = joblib.load("titanic_pipeline.joblib")
pred = model.predict(pd.DataFrame([{
    "age": 30, "fare": 7.25, "sibsp": 1, "parch": 0,
    "sex": "female", "embarked": "S", "pclass": 3,
}]))

ในขั้นตอน production คุณส่ง raw DataFrame ที่มี schema เหมือนตอน train เข้าไปได้เลย โดยไม่ต้องเขียน preprocessing ซ้ำใน service เพราะทุกขั้นถูกห่ออยู่ใน pipeline แล้ว หากใช้ FastAPI หรือ Flask serve โมเดล ขั้นตอน input validation ก็จะเรียบง่ายขึ้นมาก

หากคุณคุ้นเคยกับการรวมข้อมูลจากหลายตารางก่อนเทรน คู่มือ pandas merge, join และ concat จะช่วยให้คุณเตรียม DataFrame ก้อนเดียวที่พร้อมป้อนเข้า Pipeline ได้อย่างปลอดภัย

ข้อผิดพลาดที่พบบ่อยเมื่อใช้ Pipeline

มือใหม่มักพบปัญหาเดิมๆ เมื่อเริ่มใช้ Pipeline ครั้งแรก สรุปข้อผิดพลาดที่พบบ่อยและวิธีแก้:

• ลืม set_output("pandas"): ทำให้ ColumnTransformer คืน numpy array ที่ไม่มีชื่อคอลัมน์ ทำ debug ยาก
• ใช้ fit_transform บน test set: ทำให้ scaler/imputer เรียนรู้สถิติของ test set แก้โดยเรียก transform() เท่านั้น (Pipeline จัดการให้อยู่แล้ว)
• วาง SMOTE ใน sklearn Pipeline: ทำให้ oversampling ทำงานตอน predict ด้วย ต้องใช้ imblearn.pipeline.Pipeline แทน
• เรียก param ใน GridSearchCV ผิด syntax: ต้องเป็น step__param ไม่ใช่ step.param หรือ step:param
• ลืม handle_unknown="ignore" ที่ OneHotEncoder: production จะ error เมื่อพบ category ใหม่

หากคุณกำลังเปรียบเทียบประสิทธิภาพระหว่าง pandas และ Polars สำหรับขั้น preprocessing ขนาดใหญ่ ลองอ่าน Polars vs Pandas 2026 เพื่อตัดสินใจว่าควรเปลี่ยน backend หรือไม่ ก่อนนำข้อมูลเข้า Pipeline

สำหรับเอกสารอ้างอิงเชิงลึกของแต่ละ class ดู API ของ sklearn.pipeline.Pipeline และ หน้า release ของ scikit-learn บน GitHub ที่อัปเดตรายการ feature ใหม่ทุกรุ่น

คำถามที่พบบ่อย

Pipeline ใน scikit-learn ต่างจาก make_pipeline อย่างไร?

make_pipeline เป็น factory function ที่สร้าง Pipeline ให้โดยตั้งชื่อ step ให้อัตโนมัติเป็นชื่อคลาสตัวพิมพ์เล็ก เหมาะกับ pipeline สั้นๆ ส่วน Pipeline([(name, est), ...]) ใช้เมื่อต้องการตั้งชื่อ step เองเพื่ออ้างใน GridSearchCV หรือเข้าถึงผ่าน named_steps

ColumnTransformer กับ FeatureUnion ใช้ตอนไหน?

ใช้ ColumnTransformer เมื่อต้องการประมวลผลคนละกลุ่มคอลัมน์ด้วยขั้นตอนต่างกัน เช่น คอลัมน์ตัวเลข vs categorical ใช้ FeatureUnion เมื่อต้องการนำคอลัมน์ชุดเดียวกันไปทำหลายการแปลงแล้วต่อผลลัพธ์เข้าด้วยกัน เช่น PCA + RBF kernel จากข้อมูลตัวเดียวกัน

Pipeline ป้องกัน data leakage ได้จริงหรือไม่?

ใช่ เมื่อใช้กับ cross_val_score, cross_validate, หรือ GridSearchCV ระบบจะ fit transformer ใหม่บนแต่ละ training fold ทำให้สถิติของ validation fold ไม่รั่วเข้ามาในขั้น preprocessing แต่ถ้าคุณ fit Pipeline บนข้อมูลทั้งหมดก่อนแบ่ง split ก็จะเกิด leakage อยู่ดี

บันทึก scikit-learn Pipeline ลงไฟล์อย่างไร?

ใช้ joblib.dump(pipe, "model.joblib", compress=3) เพื่อบันทึก จากนั้นโหลดด้วย joblib.load("model.joblib") ในฝั่ง production ต้องระวังว่าเวอร์ชัน scikit-learn, NumPy, และ Python ต้องตรงกันระหว่าง train และ deploy หากต้องการ deploy ข้ามภาษาให้แปลงเป็น ONNX ด้วย skl2onnx

ใช้ Pipeline กับ XGBoost หรือ LightGBM ได้ไหม?

ได้ ทั้ง XGBoost และ LightGBM มี API ที่เข้ากันได้กับ scikit-learn (XGBClassifier, LGBMClassifier) จึงนำมาวางเป็น step สุดท้ายของ Pipeline ได้ตามปกติ และยังใช้ GridSearchCV ปรับ n_estimators, learning_rate, และ max_depth ผ่าน syntax model__n_estimators ได้