خطوط أنابيب scikit-learn في 2026: دليل عملي لبناء Pipeline احترافي مع ColumnTransformer

آخر تحديث: 2 يونيو 2026

خط أنابيب scikit-learn (sklearn.pipeline.Pipeline) هو كائن يجمع عدة خطوات معالجة وتحويل ونمذجة في كائن واحد قابل للتدريب والحفظ والاستدعاء بأمر واحد، بحيث تُطبَّق كل خطوات المعالجة المسبقة على بيانات التدريب والاختبار بنفس الطريقة تمامًا ودون تسرّب للبيانات. حسنًا، لنبدأ. في هذا الدليل العملي لإصدار scikit-learn 1.6 (مارس 2026) سنبني خط أنابيب احترافيًا من الصفر باستخدام ColumnTransformer ومحولات مخصصة، ثم نضبط المعاملات الفائقة باستخدام GridSearchCV، ونحفظ النموذج جاهزًا للإنتاج. أعترف أنني وقعت في فخ تسرّب البيانات أكثر من مرة في مشاريع سابقة قبل أن أعتمد Pipeline بشكل صارم، ومنذ ذلك الحين لم أعد أكتب أي نموذج إنتاجي بدونه.

• Pipeline يضمّن كل خطوات المعالجة والنمذجة في كائن واحد ويمنع تسرب البيانات من مجموعة الاختبار إلى التدريب.
• ColumnTransformer يطبّق محولات مختلفة على أعمدة مختلفة (رقمية، فئوية، نصية) في خطوة واحدة متوازية.
• استخدم make_pipeline وmake_column_transformer للاختصار عندما لا تحتاج لتسمية الخطوات يدويًا.
• المحولات المخصصة تُبنى عبر وراثة BaseEstimator وTransformerMixin، أو ببساطة عبر FunctionTransformer للدوال البسيطة.
• منذ scikit-learn 1.2 يمكن إخراج DataFrame من خط الأنابيب عبر set_output(transform="pandas")، وفي 1.4+ يدعم Polars أيضًا.
• احفظ النموذج بـjoblib.dump مع تثبيت إصدارات المكتبات في requirements.txt لضمان قابلية التحميل في الإنتاج.

ما هو Pipeline في scikit-learn ولماذا تحتاجه؟

خط الأنابيب في scikit-learn هو سلسلة مرتّبة من المحولات (Transformers) ينتهي اختياريًا بمُقدِّر (Estimator) واحد. كل محول يطبّق طريقتي fit وtransform، والمُقدِّر النهائي يطبّق fit وpredict. عند استدعاء pipeline.fit(X_train, y_train) تُدرَّب الخطوات بالتسلسل، ويُمرَّر خرج كل خطوة إلى مدخل الخطوة التالية.

الفائدة الجوهرية هي منع تسرب البيانات (Data Leakage). تخيّل أنك طبّقت StandardScaler على كامل مجموعة البيانات قبل التقسيم: في هذه الحالة استخدم المتوسط والانحراف المعياري لمجموعة الاختبار لحساب المعايير، مما يُعطي تقديرًا متفائلًا زائفًا للأداء. واجهت هذا الموقف بنفسي على مشروع تصنيف عملاء قبل عامين، حيث ارتفعت دقة التحقق المتقاطع بنحو 3 نقاط مئوية بسبب تسرّب بسيط في خطوة التطبيع، ولم أكتشف الأمر إلا بعد فشل النموذج في الإنتاج. مع Pipeline يُحسَب fit فقط على بيانات التدريب داخل كل طية في التقاطع المتقاطع (Cross Validation)، ثم يُطبَّق transform على بيانات الاختبار. هذا السلوك تلقائي ولا يمكن نسيانه.

إضافةً إلى ذلك، يبسّط Pipeline البحث الشبكي عن المعاملات الفائقة لخطوات متعددة في وقت واحد، ويقلّل الكود المتكرر، ويُسهّل حفظ النموذج كاملًا (المعالجة المسبقة + النموذج) في ملف واحد بـjoblib. وفقًا للوثائق الرسمية لـ scikit-learn حول التركيب، يُعتبر Pipeline الطريقة الموصى بها لإنتاج نماذج قابلة للنشر.

بناء Pipeline أساسي خطوة بخطوة

لنبدأ بمثال بسيط، تطبيع البيانات الرقمية ثم تدريب نموذج انحدار لوجستي. سنستخدم مجموعة بيانات Iris المدمجة في scikit-learn.

from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression

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

pipe = Pipeline(steps=[
    ("scaler", StandardScaler()),
    ("clf", LogisticRegression(max_iter=1000, random_state=42)),
])

pipe.fit(X_train, y_train)
print(f"الدقة على الاختبار: {pipe.score(X_test, y_test):.3f}")
# الدقة على الاختبار: 0.967

كل خطوة في steps هي زوج (اسم، كائن). الاسم يُستخدم لاحقًا للوصول إلى الخطوة عبر pipe.named_steps["scaler"] أو لتحديد المعاملات في البحث الشبكي عبر صيغة الشرطتين السفليتين scaler__with_mean. لاحظ أن pipe.predict(X_test) يطبّق StandardScaler.transform ثم LogisticRegression.predict تلقائيًا. الأمر بهذه البساطة.

معالجة أعمدة متعددة الأنواع مع ColumnTransformer

في مجموعات البيانات الواقعية تختلط الأعمدة الرقمية والفئوية والنصية، وكل نوع يحتاج معالجة مختلفة. هنا يأتي دور ColumnTransformer الذي يطبّق محولات مختلفة على مجموعات أعمدة محددة بالتوازي، ثم يدمج النتائج في مصفوفة واحدة.

مثال عملي على مجموعة بيانات تيتانيك المُبسّطة:

import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.ensemble import RandomForestClassifier

df = pd.read_csv("titanic.csv")
y = df["survived"]
X = df.drop(columns=["survived"])

numeric_features = ["age", "fare", "sibsp", "parch"]
categorical_features = ["sex", "embarked", "pclass"]

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

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

preprocessor = ColumnTransformer(transformers=[
    ("num", numeric_pipeline, numeric_features),
    ("cat", categorical_pipeline, categorical_features),
], remainder="drop", verbose_feature_names_out=False)

model = Pipeline([
    ("preprocess", preprocessor),
    ("clf", RandomForestClassifier(n_estimators=300, random_state=42)),
])

model.fit(X, y)

المعامل remainder="drop" يتخلّص من الأعمدة غير المذكورة. استخدم remainder="passthrough" لتمريرها كما هي. أما verbose_feature_names_out=False فيُبقي أسماء الأعمدة الناتجة بسيطة بدون بادئات مثل num__age. هذا الإعداد ضروري إذا كنت تريد قراءة أهمية الميزات لاحقًا.

ملاحظة مهمة: OneHotEncoder مع handle_unknown="ignore" يُنشئ متجهًا صفريًا للفئات الجديدة التي تظهر في الإنتاج ولم تكن موجودة في التدريب، وهو السلوك الأكثر أمانًا. إذا كنت ترغب في تقليل أبعاد الفئات النادرة، فإن المعامل min_frequency=10 يجمعها في فئة infrequent_sklearn واحدة منذ إصدار 1.1. لمعرفة المزيد عن أهمية تنظيف البيانات قبل النمذجة، راجع دليلنا حول التحقق من صحة البيانات في pandas باستخدام Pandera.

كتابة محول مخصص (Custom Transformer)

عاجلًا أم آجلًا ستحتاج إلى منطق معالجة لا توفّره scikit-learn جاهزًا، مثل استخراج ميزات من تواريخ أو حساب نسب بين أعمدة. هناك طريقتان: FunctionTransformer للدوال البسيطة عديمة الحالة (Stateless)، أو فئة كاملة ترث من BaseEstimator وTransformerMixin للمحولات التي تحتاج لتعلم معلمات في fit.

الطريقة 1: FunctionTransformer

import numpy as np
from sklearn.preprocessing import FunctionTransformer

log_transformer = FunctionTransformer(
    func=np.log1p,
    inverse_func=np.expm1,
    feature_names_out="one-to-one",
    validate=True,
)

pipe = Pipeline([
    ("log", log_transformer),
    ("scaler", StandardScaler()),
    ("clf", LogisticRegression()),
])

الطريقة 2: فئة محول مخصصة

from sklearn.base import BaseEstimator, TransformerMixin

class RatioFeatures(BaseEstimator, TransformerMixin):
    """يضيف عمود نسبة بين عمودين رقميين."""

    def __init__(self, numerator: str, denominator: str, eps: float = 1e-6):
        self.numerator = numerator
        self.denominator = denominator
        self.eps = eps

    def fit(self, X, y=None):
        # لا توجد معلمات للتعلم، لكن fit مطلوبة دائمًا
        self.n_features_in_ = X.shape[1]
        return self

    def transform(self, X):
        X = X.copy()
        X[f"{self.numerator}_per_{self.denominator}"] = (
            X[self.numerator] / (X[self.denominator] + self.eps)
        )
        return X

    def get_feature_names_out(self, input_features=None):
        return np.array(list(input_features) + [
            f"{self.numerator}_per_{self.denominator}"
        ])

وراثة TransformerMixin تمنحك fit_transform مجانًا، وBaseEstimator تمنحك get_params وset_params الضروريين للبحث الشبكي. لاحظ أن أي معلمة تمررها لـ__init__ يجب أن تُحفظ باسم مطابق تمامًا (self.numerator = numerator)؛ خلاف ذلك يفشل استنساخ المُقدِّر داخل cross_val_score. واجهت هذا الخطأ شخصيًا مرة عندما سمّيت السمة self._numerator بدلًا من self.numerator، وأمضيت ساعة كاملة في تتبّع رسالة خطأ غامضة. للمزيد عن العمليات الفعّالة على DataFrames، اطّلع على دليلنا حول تسريع عمليات pandas بالعمليات المتجهة.

ضبط المعاملات الفائقة مع GridSearchCV

القوة الحقيقية لـPipeline تظهر في البحث الشبكي. يمكنك ضبط معاملات أي خطوة في الأنبوب باستخدام صيغة اسم_الخطوة__اسم_المعامل:

from sklearn.model_selection import GridSearchCV

param_grid = {
    "preprocess__num__imputer__strategy": ["mean", "median"],
    "preprocess__cat__onehot__min_frequency": [None, 10, 50],
    "clf__n_estimators": [100, 300, 500],
    "clf__max_depth": [None, 10, 20],
}

grid = GridSearchCV(
    model,
    param_grid=param_grid,
    cv=5,
    scoring="roc_auc",
    n_jobs=-1,
    verbose=1,
)
grid.fit(X, y)
print(grid.best_params_)
print(f"أفضل AUC: {grid.best_score_:.4f}")

لاحظ كيف يمكنك ضبط معاملات المعالجة المسبقة (مثل استراتيجية ملء القيم المفقودة) والنموذج النهائي في نفس البحث. بدون Pipeline يصبح هذا مستحيلًا عمليًا. إذا كان فضاء البحث كبيرًا، استبدل GridSearchCV بـRandomizedSearchCV أو HalvingGridSearchCV (متاح كميزة مستقرة منذ 1.4) الذي يقلّل وقت البحث إلى النصف عبر تخصيص موارد متزايدة للمرشحين الواعدين.

إخراج DataFrame من Pipeline عبر set_output

افتراضيًا تُعيد محولات scikit-learn مصفوفات NumPy دون أسماء أعمدة، مما يعقّد تحليل الأخطاء والتفسير. منذ إصدار 1.2 (ديسمبر 2022) أصبح بإمكانك تفعيل إخراج pandas في كامل الأنبوب بسطر واحد، ومنذ 1.4 يدعم Polars أيضًا:

model.set_output(transform="pandas")
# أو
model.set_output(transform="polars")  # 1.4+

processed = model[:-1].fit_transform(X_train)  # كل الخطوات عدا المصنّف
print(type(processed))  # 
print(processed.head())

هذا يجعل تتبّع المعالجة وفحص الأعمدة الناتجة من OneHotEncoder أو المحولات المخصصة أسهل بكثير. صراحةً، أصبحت أُفعّل هذا الخيار دائمًا أثناء التطوير ثم أعيده إلى "default" عند الانتقال إلى الإنتاج للحصول على مصفوفات NumPy الأسرع. وفقًا لسجل تغييرات scikit-learn، يعمل فريق التطوير على دعم خرج Arrow أيضًا في الإصدارات القادمة.

حفظ النموذج وتحميله في الإنتاج

الميزة الأخيرة الحاسمة هي القدرة على حفظ Pipeline كاملًا (المعالجة + النموذج) في ملف واحد:

import joblib

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

# لاحقًا في خدمة الإنتاج:
loaded = joblib.load("titanic_pipeline.joblib")
predictions = loaded.predict(new_data)

المعامل compress=3 يقلل حجم الملف بشكل ملحوظ مع تأثير طفيف على وقت التحميل. للنماذج الضخمة (مئات الميجابايت) جرّب compress=("lz4", 3) الأسرع. إذا كنت ترغب في تقليل بصمة الذاكرة لـDataFrames المستخدمة في التدريب، راجع دليلنا حول تحسين استهلاك الذاكرة في pandas.

أخطاء شائعة وكيفية تجنبها

إليك أربعة أخطاء أراها بشكل متكرر في مراجعات الكود، مع الحلول:

1. تطبيق fit_transform على بيانات الاختبار: هذا تسريب بيانات صريح. استخدم transform فقط على الاختبار. مع Pipeline لا يحدث ذلك أبدًا لأنك تستدعي pipeline.predict(X_test) مباشرة.
2. وضع SMOTE داخل Pipeline عادي: SMOTE من مكتبة imbalanced-learn يطبّق إعادة العيّنة على التدريب فقط، لكن Pipeline القياسي يطبّقها أيضًا في التنبؤ ويُفسد النتائج. استخدم imblearn.pipeline.Pipeline بدلًا منه.
3. نسيان random_state: أي مكوّن عشوائي (RandomForest، GridSearchCV، train_test_split) يحتاج بذرة ثابتة لإعادة الإنتاج. اضبطها على القيمة نفسها (42 مثلًا) في كل المكونات.
4. اختراق التغليف عبر تعديل X مباشرة: في المحولات المخصصة استنسخ DataFrame بـX.copy() قبل أي تعديل، وإلا ستتفاجأ بأخطاء غامضة عند إعادة تشغيل خلية Jupyter.

الأسئلة الشائعة

ما الفرق بين Pipeline و make_pipeline في scikit-learn؟

Pipeline يتطلب قائمة من الأزواج (اسم، كائن) ويمنحك تحكمًا كاملًا في أسماء الخطوات، وهو الخيار المفضل عندما تحتاج لاستهداف خطوة معينة في GridSearchCV. أما make_pipeline فيستقبل الكائنات فقط ويولّد الأسماء تلقائيًا من اسم الصنف بحروف صغيرة، وهو أقصر للسكربتات السريعة.

هل يمكن استخدام ColumnTransformer مع NumPy arrays بدون أسماء أعمدة؟

نعم، بدلًا من تمرير قائمة أسماء أعمدة مرّر فهارس صحيحة مثل [0, 1, 2] أو شرائح slice(0, 5). يمكنك أيضًا استخدام make_column_selector(dtype_include=np.number) لاختيار الأعمدة ديناميكيًا بناءً على نوع البيانات.

كيف أحصل على أسماء الميزات بعد المعالجة المسبقة؟

استدعِ pipeline.named_steps["preprocess"].get_feature_names_out() بعد fit. هذه الطريقة مدعومة في كل محولات scikit-learn الحديثة وفي محولاتك المخصصة إذا أضفت طريقة get_feature_names_out. بديلًا، فعّل set_output(transform="pandas") واحصل على DataFrame مع أسماء الأعمدة مباشرة.

هل Pipeline يعمل مع التعلم العميق أو نماذج PyTorch؟

scikit-learn Pipeline مصمم للمُقدّرات التي تتبع واجهة scikit-learn (fit, predict). للتعلم العميق استخدم مكتبة skorch التي تغلّف نماذج PyTorch بواجهة متوافقة مع Pipeline و GridSearchCV، أو scikeras لنماذج Keras.

ما البديل لـ joblib لحفظ النماذج بشكل أكثر أمانًا؟

مكتبة skops الرسمية من scikit-learn تقدم تنسيقًا أكثر أمانًا (لا ينفّذ كودًا عند التحميل) ومتوافقًا أفضل مع تغييرات الإصدارات. بديل آخر هو تحويل النموذج إلى ONNX عبر skl2onnx لتشغيله على أي منصة دون الحاجة لـ Python في الإنتاج.