انتقل إلى المحتوى الرئيسي

الوحدة 9 — SageMaker Pipelines وسجلّ النماذج

الوحدات السابقة عالجت كلّ خطوة على حدة: بيانات، تدريب، ضبط، نشر، دفعة. في هذه الوحدة نربط الكلّ في سلسلة قابلة للتكرار تعمل بلمسة زر، ويعرف سجلّ النماذج (Model Registry) أيّ نسخة موافقٌ عليها لإرسالها إلى الإنتاج. هذه هي الأتمتة التي تصنع MLOps الفعليّة على AWS.

لماذا خطوط الإنتاج​

بعد أسبوعين على الخيط الأحمر، تصبح الحلقة اليدويّة عبئًا:

  1. تشغيل مهمّة معالجة على أحدث لقطة.
  2. تشغيل تدريب.
  3. قراءة المقياس.
  4. تشغيل ضبط تلقائيّ.
  5. تشغيل تقييم على مجموعة الاختبار.
  6. القرار: هل نستحقّ إرسال هذا إلى الإنتاج؟

هذا التسلسل يتكرّر كلّ شهر. كتابته يدويًّا يعني نسيان خطوة، أو استعمال بيانات مختلفة، أو نشر نسخة لم تُقيَّم فعلًا. SageMaker Pipelines يعرّف هذا التسلسل كـDAG (رسم بيانيّ اتّجاهيّ لا دوريّ) قابل للاحتجاز والتشغيل بأمر واحد.

الخطوات الأربع الأشيع​

خطوط SageMaker مؤلَّفة من خطوات، كلّ منها لها نوع محدّد:

  • ProcessingStep: تشغيل حاوية معالجة (scikit-learn، Spark، أو مخصّصة).
  • TrainingStep: تشغيل مهمّة تدريب.
  • ConditionStep: تفريع مبنيّ على قيمة (مقياس المصادقة).
  • RegisterModel أو ModelStep: تسجيل النموذج في السجلّ.

يمكن إضافة TuningStep وTransformStep وClarifyCheckStep. المبدأ ثابت: كلّ خطوة تصف دخلها ومخرَجها، ويستنتج SageMaker الترتيب من التبعيّة.

خطّ مبسّط لخيطنا الأحمر​

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.workflow.parameters import ParameterString, ParameterFloat

# 1. Paramètres
raw_data = ParameterString(name="RawDataUri", default_value=f"s3://{bucket}/churn/raw/latest/")
f1_threshold = ParameterFloat(name="F1Threshold", default_value=0.72)

# 2. Étape de préparation
prep_step = ProcessingStep(name="Prepare", processor=sklearn_processor, ...)

# 3. Étape d'entraînement
train_step = TrainingStep(name="Train", estimator=sk_estimator,
                          inputs={"train": prep_step.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri})

# 4. Étape d'évaluation
eval_step = ProcessingStep(name="Evaluate", processor=eval_processor,
                           inputs=[..., train_step.properties.ModelArtifacts.S3ModelArtifacts])

# 5. Condition sur le F1
register_step = ModelStep(name="Register", ...)
cond = ConditionGreaterThanOrEqualTo(
    left=JsonGet(step_name=eval_step.name, property_file=evaluation_report, json_path="metrics.f1"),
    right=f1_threshold,
)
condition_step = ConditionStep(name="F1Gate", conditions=[cond], if_steps=[register_step], else_steps=[])

# 6. Composition
pipeline = Pipeline(
    name="churn-pipeline",
    parameters=[raw_data, f1_threshold],
    steps=[prep_step, train_step, eval_step, condition_step],
)
pipeline.upsert(role_arn=role)
execution = pipeline.start(parameters={"F1Threshold": 0.75})

الشرط: البوّابة التي تحمي الإنتاج​

ConditionStep هو أهمّ سطر في الخطّ. بغيره، أيّ نموذج مدرَّب يُسجَّل، حتّى لو كان دون النموذج السابق. مع الشرط:

  • إذا F1 ≥ 0.72 → التسجيل، والانتقال إلى مراجعة يدويّة.
  • وإلّا → توقّف صامت، لا تسجيل، لا نشر.

هذا الشرط لا يعوّض إشراف الإنسان، بل يفلتر ما هو دون الاعتبار قبل أن يصل إلى فريق المراجعة. القاعدة العمليّة: عتبة قليلة تحت أفضل نموذج حاليّ (مثلًا 0.02 أقلّ)، لتفادي رفض تحسينات صغيرة قد تكون واقعيّة.

سجلّ النماذج (Model Registry)​

RegisterModel يضيف النموذج إلى مجموعة نماذج (Model Package Group)، مع حالة المصادقة:

  • PendingManualApproval (افتراضيّة): يظهر في Studio، ينتظر مراجعًا.
  • Approved: قابل للنشر إلى الإنتاج.
  • Rejected: محفوظ للأرشيف، ممنوع من النشر.

الفرد المصادق (Data Scientist Lead، مثلًا) يفتح Studio → Model Registry → يقارن مقاييس النسخ، يقرأ التقرير المرفق (JSON مخرَج من evaluation.py)، ثمّ يوافق أو يرفض. المصادقة تُنشئ حدثًا في EventBridge يمكن ربطه بخطّ نشر آخر يُشغّل تلقائيًّا.

المزيّة العمليّة الكبرى: تاريخ النماذج قابل للاسترجاع. عند حادثة إنتاج تتّهم فيها نسخة جديدة، تنشر النسخة السابقة بأمرين، بلا إعادة تدريب.

التفعيل: EventBridge + Lambda​

يعمل الخطّ يدويًّا بـpipeline.start()، لكنّ الإنتاج يحتاج تفعيلًا بلا مطوّر أمام الشاشة:

  • جدول شهريّ: قاعدة CloudWatch/EventBridge تشغّل Lambda بسيطة كلّ يوم أوّل، تنادي pipeline.start().
  • بحدث بيانات: كلّ وصول ملفّ جديد إلى s3://.../raw/ يشغّل الخطّ (S3 Event → Lambda).
  • بحدث مصادقة: كلّ تحديث لنموذج إلى Approved يشغّل خطّ نشر منفصل.
# Extrait Lambda de démarrage mensuel
import boto3
def handler(event, context):
    client = boto3.client("sagemaker")
    client.start_pipeline_execution(PipelineName="churn-pipeline")

المصادر (Lineage) تلقائيًّا​

كلّ تشغيل يترك أثرًا كاملًا في SageMaker Lineage: أيّ بيانات دخلت، أيّ سكربت شُغِّل، أيّ صورة استُعملت، أيّ نموذج نتج، أيّ توقّعات كتبت. عند سؤال المدقّق «كيف حُسب توقّع تسرّب هذا العميل قبل ثمانية أشهر؟»، الإجابة موجودة، دون تنقيب في السجلّات.

هذه هي القيمة الحقيقيّة لخطّ إنتاج على Pipelines، ليست فقط الأتمتة بل التتبّع الدائم الذي يسمح بالمراجعة والامتثال.

ابدأ صغيرًا

لا تحاول بناء خطّ يتضمّن الاثنتي عشرة خطوة مرّة واحدة. ابدأ بثلاث خطوات (تحضير، تدريب، تقييم)، جرّبه، تأكّد من قراءة المقياس، ثمّ أضف الشرط، ثمّ التسجيل، ثمّ النشر. كلّ خطوة إضافيّة معقّدة لتصحيح كوّة في نصف ساعة أسهل من تصحيح خطّ كامل ينهار من غير سبب واضح.

الخلاصة​

  • SageMaker Pipelines يعرّف السلسلة كـDAG قابل للتكرار، ويستنتج الترتيب من تبعيّات المخرَجات.
  • ConditionStep يحمي الإنتاج بشرط على المقياس؛ لا يعوّض المراجعة البشريّة لكنّه يفلتر السيّئ آليًّا.
  • سجلّ النماذج يحفظ تاريخ النسخ مع حالة المصادقة، ويتيح استرجاع نسخة سابقة في أمرين.
  • التفعيل بجدول أو بحدث بيانات أو بمصادقة يفصل التشغيل عن حضور المطوّر.

الوحدة التالية والأخيرة: كيف نراقب النموذج والفاتورة معًا في الإنتاج.