ایمیزون سیج میکر پائپ لائنز اور ایمیزون سیج میکر SDK کے لیے نئی خصوصیات

ایمیزون سیج میکر پائپ لائنز ڈیٹا سائنسدانوں اور مشین لرننگ (ML) انجینئرز کو تربیتی ورک فلو کو خودکار کرنے کی اجازت دیتا ہے، جو آپ کو تیز تجربہ اور ماڈل کی دوبارہ تربیت کے لیے ماڈل کی ترقی کے مراحل کو ترتیب دینے کے لیے دوبارہ قابل عمل عمل بنانے میں مدد کرتا ہے۔ آپ پورے ماڈل بلڈ ورک فلو کو خودکار کر سکتے ہیں، بشمول ڈیٹا کی تیاری، فیچر انجینئرنگ، ماڈل ٹریننگ، ماڈل ٹیوننگ، اور ماڈل کی توثیق، اور اسے ماڈل رجسٹری میں کیٹلاگ کر سکتے ہیں۔ آپ پائپ لائنوں کو باقاعدہ وقفوں پر خود بخود چلانے کے لیے ترتیب دے سکتے ہیں یا جب کچھ واقعات شروع ہوتے ہیں، یا آپ ضرورت کے مطابق انہیں دستی طور پر چلا سکتے ہیں۔

اس پوسٹ میں، ہم اس میں سے کچھ اضافہ کو اجاگر کرتے ہیں۔ ایمیزون سیج میکر SDK اور Amazon SageMaker Pipelines کی نئی خصوصیات متعارف کروائیں جو ML پریکٹیشنرز کے لیے ML ماڈلز کی تعمیر اور تربیت کو آسان بناتی ہیں۔

پائپ لائنز اپنے ڈویلپر کے تجربے میں جدت لاتی رہتی ہیں، اور ان حالیہ ریلیزز کے ساتھ، اب آپ سروس کو زیادہ حسب ضرورت طریقے سے استعمال کر سکتے ہیں:

• 2.99.0, 2.101.1, 2.102.0, 2.104.0 - اپ ڈیٹ شدہ دستاویزات پر PipelineVariable تخمینہ لگانے والے، پروسیسر، ٹونر، ٹرانسفارمر، اور ماڈل بیس کلاسز، ایمیزون ماڈلز، اور فریم ورک ماڈلز کے لیے استعمال۔ تخمینوں اور پروسیسرز کے تمام ذیلی طبقات کو سپورٹ کرنے کے لیے SDK کے نئے ورژن کے ساتھ آنے والی اضافی تبدیلیاں ہوں گی۔
• 2.90.0 - کی دستیابی ماڈل سٹیپ مربوط ماڈل وسائل کی تخلیق اور رجسٹریشن کے کاموں کے لیے۔
• 2.88.2 - کی دستیابی پائپ لائن سیشن SageMaker اداروں اور وسائل کے ساتھ منظم تعامل کے لیے۔
• 2.88.2 - کے لیے ذیلی طبقے کی مطابقت ورک فلو پائپ لائن کام کے مراحل تاکہ آپ جاب کے تجریدات بنا سکیں اور پروسیسنگ، ٹریننگ، ٹرانسفارم اور ٹیوننگ جابز کو ترتیب اور چلا سکیں جیسا کہ آپ پائپ لائن کے بغیر کرتے ہیں۔
• 2.76.0 - کی دستیابی ناکام قدم ناکامی کی حالت کے ساتھ پائپ لائن کو مشروط طور پر روکنا۔

اس پوسٹ میں، ہم آپ کو نمونے کے ڈیٹاسیٹ کا استعمال کرتے ہوئے ایک ورک فلو سے گزرتے ہیں جس میں ماڈل کی تعمیر اور تعیناتی پر فوکس کیا جاتا ہے تاکہ یہ ظاہر کیا جا سکے کہ پائپ لائنز کی نئی خصوصیات کو کیسے نافذ کیا جائے۔ آخر تک، آپ کے پاس ان نئی خصوصیات کو کامیابی سے استعمال کرنے اور اپنے ML ورک بوجھ کو آسان بنانے کے لیے کافی معلومات ہونی چاہیے۔

خصوصیات کا جائزہ

پائپ لائنز درج ذیل نئی خصوصیات پیش کرتی ہیں:

• پائپ لائن متغیر تشریح - کچھ طریقہ کار پیرامیٹرز متعدد ان پٹ اقسام کو قبول کرتے ہیں، بشمول PipelineVariables، اور یہ واضح کرنے کے لیے اضافی دستاویزات شامل کی گئی ہیں کہ کہاں PipelineVariables SageMaker SDK دستاویزات کے تازہ ترین مستحکم ورژن اور فنکشنز کے init دستخط دونوں میں تعاون یافتہ ہیں۔ مثال کے طور پر، درج ذیل TensorFlow تخمینہ کنندہ میں، init دستخط اب یہ ظاہر کرتا ہے۔ model_dir اور image_uri حمایت PipelineVariables، جبکہ دوسرے پیرامیٹرز ایسا نہیں کرتے ہیں۔ مزید معلومات کے لیے رجوع کریں۔ ٹینسر فلو کا تخمینہ لگانے والا.
  • اس سے پہلے:

    TensorFlow(
        py_version=None,
        framework_version=None,
        model_dir=None,
        image_uri=None,
        distribution=None,
        **kwargs,
    )

  • کے بعد:

    TensorFlow(
        py_version: Union[str, NoneType] = None,
        framework_version: Union[str, NoneType] = None,
        model_dir: Union[str, sagemaker.workflow.entities.PipelineVariable, NoneType] = None,
        image_uri: Union[str, sagemaker.workflow.entities.PipelineVariable, NoneType] = None,
        distribution: Union[Dict[str, str], NoneType] = None,
        compiler_config: Union[sagemaker.tensorflow.training_compiler.config.TrainingCompilerConfig, NoneType] = None,
        **kwargs,
    )

• پائپ لائن سیشن - پائپ لائن سیشن ایک نیا تصور ہے جو SageMaker SDK میں اتحاد لانے کے لیے متعارف کرایا گیا ہے اور پائپ لائن کے وسائل کی سست شروعات کو متعارف کرایا گیا ہے (رن کالز کیپچر ہو جاتی ہیں لیکن پائپ لائن بننے اور چلانے تک نہیں چلتی ہیں)۔ دی PipelineSession سیاق و سباق وراثت میں ملتا ہے۔ SageMakerSession اور آپ کے لیے SageMaker کے دیگر اداروں اور وسائل کے ساتھ تعامل کرنے کے لیے آسان طریقے نافذ کرتا ہے، جیسے کہ تربیتی جابز، اینڈ پوائنٹس، اور ان پٹ ڈیٹا سیٹس ایمیزون سادہ اسٹوریج سروس (ایمیزون S3)۔
• ورک فلو پائپ لائن کام کے مراحل کے ساتھ ذیلی طبقے کی مطابقت - اب آپ جاب کے خلاصے بنا سکتے ہیں اور پروسیسنگ، ٹریننگ، ٹرانسفارم اور ٹیوننگ جابز کو ترتیب اور چلا سکتے ہیں جیسا کہ آپ پائپ لائن کے بغیر کرتے ہیں۔
  • مثال کے طور پر، کے ساتھ ایک پروسیسنگ قدم بنانا SKLearnProcessor پہلے درج ذیل کی ضرورت تھی:

        sklearn_processor = SKLearnProcessor(
            framework_version=framework_version,
            instance_type=processing_instance_type,
            instance_count=processing_instance_count,
            sagemaker_session=sagemaker_session, #sagemaker_session would be passed as an argument
            role=role,
        )
        step_process = ProcessingStep(
            name="{pipeline-name}-process",
            processor=sklearn_processor,
            inputs=[
              ProcessingInput(source=input_data, destination="/opt/ml/processing/input"),  
            ],
            outputs=[
                ProcessingOutput(output_name="train", source="/opt/ml/processing/train"),
                ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"),
                ProcessingOutput(output_name="test", source="/opt/ml/processing/test")
            ],
            code=f"code/preprocess.py",
        )

  • جیسا کہ ہم پچھلے کوڈ میں دیکھتے ہیں، ProcessingStep بنیادی طور پر وہی پری پروسیسنگ منطق کرنے کی ضرورت ہے۔ .run، کام شروع کرنے کے لیے API کال شروع کیے بغیر۔ لیکن ذیلی طبقے کی مطابقت کے ساتھ اب ورک فلو پائپ لائن جاب کے اقدامات کے ساتھ، ہم اعلان کرتے ہیں۔ step_args آرگومینٹ جو .run کے ساتھ پری پروسیسنگ منطق لیتا ہے تاکہ آپ ایک جاب خلاصہ بناسکیں اور اسے ترتیب دے سکیں جیسا کہ آپ اسے پائپ لائنز کے بغیر استعمال کریں گے۔ ہم بھی پاس کرتے ہیں۔ pipeline_session، جو ایک ہے PipelineSession اعتراض، بجائے sagemaker_session اس بات کو یقینی بنانے کے لیے کہ رن کالز پکڑی گئی ہیں لیکن اس وقت تک کال نہیں کی جائیں گی جب تک کہ پائپ لائن نہ بن جائے اور چلائی جائے۔ درج ذیل کوڈ دیکھیں:

    sklearn_processor = SKLearnProcessor(
        framework_version=framework_version,
        instance_type=processing_instance_type,
        instance_count=processing_instance_count,
        sagemaker_session=pipeline_session,#pipeline_session would be passed in as argument
        role=role,
    )
    
    processor_args = sklearn_processor.run(
        inputs=[
          ProcessingInput(source=input_data, destination="/opt/ml/processing/input"),  
        ],
        outputs=[
            ProcessingOutput(output_name="train", source="/opt/ml/processing/train"),
            ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"),
            ProcessingOutput(output_name="test", source="/opt/ml/processing/test")
        ],
        code=f"code/preprocess.py",
    )
    step_process = ProcessingStep(name="{pipeline-name}-process", step_args=processor_args)

• ماڈل مرحلہ (ماڈل کی تخلیق اور رجسٹریشن کے مراحل کے ساتھ ایک ہموار طریقہ) -پائپ لائنز سیج میکر ماڈلز کے ساتھ ضم کرنے کے لیے دو قدمی اقسام پیش کرتی ہیں: CreateModelStep اور RegisterModel. اب آپ صرف استعمال کرکے دونوں کو حاصل کرسکتے ہیں۔ ModelStep قسم نوٹ کریں کہ a PipelineSession اس کو حاصل کرنے کے لئے ضروری ہے. یہ پائپ لائن کے مراحل اور SDK کے درمیان مماثلت لاتا ہے۔
  • اس سے پہلے:

    step_register = RegisterModel(
            name="ChurnRegisterModel",
            estimator=xgb_custom_estimator,
            model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
            content_types=["text/csv"],
            response_types=["text/csv"],
            inference_instances=["ml.t2.medium", "ml.m5.large"],
            transform_instances=["ml.m5.large"],
            model_package_group_name=model_package_group_name,
            approval_status=model_approval_status,
            model_metrics=model_metrics,
    )

• • کے بعد:

    register_args = model.register(
        content_types=["text/csv"],
        response_types=["text/csv"],
        inference_instances=["ml.t2.medium", "ml.m5.xlarge"],
        transform_instances=["ml.m5.xlarge"],
        model_package_group_name=model_package_group_name,
        approval_status=model_approval_status,
        model_metrics=model_metrics,
    )
    step_register = ModelStep(name="ChurnRegisterModel", step_args=register_args)

• ناکامی کا مرحلہ (پائپ لائن چلانے کا مشروط روک) - FailStep اگر کوئی شرط پوری ہو جاتی ہے تو پائپ لائن کو ناکامی کی حالت کے ساتھ روکنے کی اجازت دیتا ہے، جیسے کہ اگر ماڈل کا سکور ایک خاص حد سے نیچے ہے۔

حل جائزہ

اس حل میں، آپ کا انٹری پوائنٹ ہے۔ ایمیزون سیج میکر اسٹوڈیو تیز رفتار تجربات کے لیے مربوط ترقیاتی ماحول (IDE)۔ اسٹوڈیو ایک ایسا ماحول پیش کرتا ہے جس سے پائپ لائنز کے اختتام تک کے تجربے کو منظم کیا جا سکے۔ سٹوڈیو کے ساتھ، آپ کو بائی پاس کر سکتے ہیں AWS مینجمنٹ کنسول آپ کے پورے ورک فلو مینجمنٹ کے لیے۔ اسٹوڈیو کے اندر سے پائپ لائنز کے انتظام کے بارے میں مزید معلومات کے لیے، رجوع کریں۔ سیج میکر اسٹوڈیو میں سیج میکر پائپ لائنز دیکھیں، ٹریک کریں اور ان پر عمل کریں۔.

مندرجہ ذیل خاکہ ML ورک فلو کے اعلیٰ سطحی فن تعمیر کو نئی خصوصیات کا استعمال کرتے ہوئے تربیت دینے اور نتائج پیدا کرنے کے مختلف مراحل کے ساتھ واضح کرتا ہے۔

پائپ لائن میں درج ذیل مراحل شامل ہیں:

1. مطلوبہ خصوصیات بنانے کے لیے ڈیٹا کو پہلے سے پروسیس کریں اور ڈیٹا کو ٹرین، توثیق اور ٹیسٹ ڈیٹا سیٹس میں تقسیم کریں۔
2. SageMaker XGBoost فریم ورک کے ساتھ تربیتی جاب بنائیں۔
3. ٹیسٹ ڈیٹاسیٹ کا استعمال کرتے ہوئے تربیت یافتہ ماڈل کا اندازہ لگائیں۔
4. چیک کریں کہ آیا AUC سکور پہلے سے طے شدہ حد سے اوپر ہے۔
   • اگر AUC سکور حد سے کم ہے تو پائپ لائن کو رن روکیں اور اسے ناکام کے طور پر نشان زد کریں۔
   • اگر AUC سکور حد سے زیادہ ہے، تو SageMaker ماڈل بنائیں اور اسے SageMaker ماڈل رجسٹری میں رجسٹر کریں۔
5. پچھلے مرحلے میں بنائے گئے ماڈل کا استعمال کرتے ہوئے دیئے گئے ڈیٹاسیٹ پر بیچ ٹرانسفارم لگائیں۔

شرائط

اس پوسٹ کے ساتھ پیروی کرنے کے لیے، آپ کو AWS اکاؤنٹ کی ضرورت ہے۔ اسٹوڈیو ڈومین.

پائپ لائنز براہ راست SageMaker اداروں اور وسائل کے ساتھ مربوط ہیں، لہذا آپ کو کسی دوسری AWS خدمات کے ساتھ تعامل کرنے کی ضرورت نہیں ہے۔ آپ کو کسی بھی وسائل کا انتظام کرنے کی بھی ضرورت نہیں ہے کیونکہ یہ ایک مکمل طور پر منظم سروس ہے، جس کا مطلب ہے کہ یہ آپ کے لیے وسائل تخلیق اور ان کا نظم کرتی ہے۔ SageMaker کے مختلف اجزاء کے بارے میں مزید معلومات کے لیے جو اسٹوڈیو کے مربوط اجزاء کے ساتھ اسٹینڈ ایلون Python APIs ہیں، دیکھیں سیج میکر پروڈکٹ پیج.

شروع کرنے سے پہلے، SageMaker SDK ورژن>= 2.104.0 اور xlrd>=1.0.0 کو اسٹوڈیو نوٹ بک کے اندر درج ذیل کوڈ کا ٹکڑا استعمال کرتے ہوئے انسٹال کریں:

print(sagemaker.__version__)
import sys
!{sys.executable} -m pip install "sagemaker>=2.104.0"
!{sys.executable} -m pip install "xlrd >=1.0.0"
 
import sagemaker

ایم ایل ورک فلو

اس پوسٹ کے لیے، آپ درج ذیل اجزاء استعمال کرتے ہیں:

• ڈیٹا کی تیاری
  • سیج میکر پروسیسنگ - سیج میکر پروسیسنگ ایک مکمل طور پر منظم سروس ہے جو آپ کو ایم ایل ورک بوجھ کے لیے حسب ضرورت ڈیٹا ٹرانسفارمیشن اور فیچر انجینئرنگ چلانے کی اجازت دیتی ہے۔
• ماڈل بلڈنگ
• ماڈل ٹریننگ اور تشخیص
  • ایک کلک کی تربیت - سیج میکر نے تربیت کی خصوصیت تقسیم کی۔ SageMaker ڈیٹا کی ہم آہنگی اور ماڈل کے ہم آہنگی کے لیے تقسیم شدہ تربیتی لائبریریاں فراہم کرتا ہے۔ لائبریریاں سیج میکر کے تربیتی ماحول کے لیے بہتر بنائی گئی ہیں، آپ کی تقسیم شدہ تربیتی ملازمتوں کو سیج میکر کے مطابق ڈھالنے میں مدد کریں، اور تربیت کی رفتار اور تھرو پٹ کو بہتر بنائیں۔
  • سیج میکر کے تجربات - تجربات SageMaker کی ایک صلاحیت ہے جو آپ کو اپنے ML تکرار کو منظم کرنے، ٹریک کرنے، موازنہ کرنے اور جانچنے دیتی ہے۔
  • سیج میکر بیچ ٹرانسفارم - بیچ ٹرانسفارم یا آف لائن اسکورنگ SageMaker میں ایک منظم سروس ہے جو آپ کو اپنے ML ماڈلز کا استعمال کرتے ہوئے ایک بڑے ڈیٹاسیٹ پر پیش گوئی کرنے دیتی ہے۔
• ورک فلو آرکیسٹریشن

سیج میکر پائپ لائن ایک دوسرے سے جڑے ہوئے اقدامات کا ایک سلسلہ ہے جس کی وضاحت JSON پائپ لائن کی تعریف کے ذریعہ کی گئی ہے۔ یہ ڈائریکٹڈ ایسکلک گراف (DAG) کا استعمال کرتے ہوئے ایک پائپ لائن کو انکوڈ کرتا ہے۔ ڈی اے جی پائپ لائن کے ہر مرحلے کی ضروریات اور ان کے درمیان تعلقات کے بارے میں معلومات فراہم کرتا ہے، اور اس کی ساخت کا تعین مراحل کے درمیان ڈیٹا کے انحصار سے کیا جاتا ہے۔ یہ انحصار اس وقت پیدا ہوتا ہے جب ایک قدم کے آؤٹ پٹ کی خصوصیات کو دوسرے قدم پر ان پٹ کے طور پر منتقل کیا جاتا ہے۔

مندرجہ ذیل خاکہ سیج میکر پائپ لائن کے مختلف مراحل کی وضاحت کرتا ہے (مچھن کی پیشن گوئی کے استعمال کے کیس کے لیے) جہاں سیج میکر کے ذریعے قدموں کے درمیان کنکشن کا اندازہ ان پٹس اور آؤٹ پٹس کی بنیاد پر لگایا گیا ہے جو قدم کی تعریفوں سے متعین ہیں۔

اگلے حصے پائپ لائن کے ہر مرحلے کو بنانے اور ایک بار بننے کے بعد پوری پائپ لائن کو چلانے کے ذریعے چلتے ہیں۔

پروجیکٹ کا ڈھانچہ

آئیے پروجیکٹ کی ساخت کے ساتھ شروع کریں:

• /sm-pipelines-end-to-end-مثال - پروجیکٹ کا نام
  • / ڈیٹا - ڈیٹاسیٹس
  • /پائپ لائنز - پائپ لائن کے اجزاء کے لیے کوڈ فائلیں۔
    • /customerchurn
      • preprocess.py
      • evaluate.py
  • sagemaker-pipelines-project.ipynb - پائپ لائنز کی نئی خصوصیات کا استعمال کرتے ہوئے ماڈلنگ ورک فلو کے ذریعے چلنے والی ایک نوٹ بک

ڈیٹاسیٹ ڈاؤن لوڈ کریں۔

اس پوسٹ کے ساتھ پیروی کرنے کے لیے، آپ کو ڈاؤن لوڈ اور محفوظ کرنے کی ضرورت ہے۔ نمونہ ڈیٹاسیٹ پروجیکٹ ہوم ڈائرکٹری کے اندر ڈیٹا فولڈر کے تحت، جو فائل کو محفوظ کرتا ہے۔ ایمیزون لچکدار فائل سسٹم (ایمیزون ای ایف ایس) اسٹوڈیو ماحول میں۔

پائپ لائن کے اجزاء بنائیں

اب آپ پائپ لائن کے اجزاء بنانے کے لیے تیار ہیں۔

بیانات درآمد کریں اور پیرامیٹرز اور مستقل کا اعلان کریں۔

نامی ایک اسٹوڈیو نوٹ بک بنائیں sagemaker-pipelines-project.ipynb پروجیکٹ ہوم ڈائرکٹری کے اندر۔ سیل میں درج ذیل کوڈ بلاک درج کریں، اور SageMaker اور S3 کلائنٹ اشیاء کو ترتیب دینے کے لیے سیل کو چلائیں، تخلیق کریں PipelineSession، اور پہلے سے طے شدہ بالٹی کا استعمال کرتے ہوئے S3 بالٹی کا مقام مرتب کریں جو SageMaker سیشن کے ساتھ آتا ہے:

import boto3
import pandas as pd
import sagemaker
from sagemaker.workflow.pipeline_context import PipelineSession
 
s3_client = boto3.resource('s3')
pipeline_name = f"ChurnModelPipeline"
sagemaker_session = sagemaker.session.Session()
region = sagemaker_session.boto_region_name
role = sagemaker.get_execution_role()
pipeline_session = PipelineSession()
default_bucket = sagemaker_session.default_bucket()
model_package_group_name = f"ChurnModelPackageGroup"

پائپ لائنز پیرامیٹرائزیشن کو سپورٹ کرتی ہیں، جو آپ کو اپنے پائپ لائن کوڈ کو تبدیل کیے بغیر رن ٹائم پر ان پٹ پیرامیٹرز کی وضاحت کرنے کی اجازت دیتی ہے۔ آپ کے تحت دستیاب ماڈیولز استعمال کر سکتے ہیں۔ sagemaker.workflow.parameters ماڈیول، جیسے ParameterInteger, ParameterFloat، اور ParameterStringمختلف ڈیٹا کی اقسام کے پائپ لائن پیرامیٹرز کی وضاحت کرنے کے لیے۔ متعدد ان پٹ پیرامیٹرز کو ترتیب دینے کے لیے درج ذیل کوڈ کو چلائیں:

from sagemaker.workflow.parameters import (
    ParameterInteger,
    ParameterString,
    ParameterFloat,
)
auc_score_threshold = 0.75
base_job_prefix = "churn-example"
model_package_group_name = "churn-job-model-packages"
batch_data = "s3://{}/data/batch/batch.csv".format(default_bucket)

processing_instance_count = ParameterInteger(
    name="ProcessingInstanceCount",
    default_value=1
)
processing_instance_type = ParameterString(
    name="ProcessingInstanceType",
    default_value="ml.m5.xlarge"
)
training_instance_type = ParameterString(
    name="TrainingInstanceType",
    default_value="ml.m5.xlarge"
)
input_data = ParameterString(
    name="InputData",
    default_value="s3://{}/data/storedata_total.csv".format(default_bucket),
)

model_approval_status = ParameterString(
    name="ModelApprovalStatus", default_value="PendingManualApproval"
)

بیچ ڈیٹاسیٹ تیار کریں۔

بیچ ڈیٹاسیٹ تیار کریں، جسے آپ بعد میں بیچ ٹرانسفارم مرحلے میں استعمال کرتے ہیں:

def preprocess_batch_data(file_path):
    df = pd.read_csv(file_path)
    ## Convert to datetime columns
    df["firstorder"]=pd.to_datetime(df["firstorder"],errors='coerce')
    df["lastorder"] = pd.to_datetime(df["lastorder"],errors='coerce')
    ## Drop Rows with null values
    df = df.dropna()
    ## Create Column which gives the days between the last order and the first order
    df["first_last_days_diff"] = (df['lastorder']-df['firstorder']).dt.days
    ## Create Column which gives the days between when the customer record was created and the first order
    df['created'] = pd.to_datetime(df['created'])
    df['created_first_days_diff']=(df['created']-df['firstorder']).dt.days
    ## Drop Columns
    df.drop(['custid','created','firstorder','lastorder'],axis=1,inplace=True)
    ## Apply one hot encoding on favday and city columns
    df = pd.get_dummies(df,prefix=['favday','city'],columns=['favday','city'])
    return df
    
# convert the store_data file into csv format
store_data = pd.read_excel("data/storedata_total.xlsx")
store_data.to_csv("data/storedata_total.csv")
 
# preprocess batch data and save into the data folder
batch_data = preprocess_batch_data("data/storedata_total.csv")
batch_data.pop("retained")
batch_sample = batch_data.sample(frac=0.2)
pd.DataFrame(batch_sample).to_csv("data/batch.csv",header=False,index=False)

S3 بالٹی میں ڈیٹا اپ لوڈ کریں۔

ایمیزون S3 پر ڈیٹا سیٹ اپ لوڈ کریں:

s3_client.Bucket(default_bucket).upload_file("data/batch.csv","data/batch/batch.csv")
s3_client.Bucket(default_bucket).upload_file("data/storedata_total.csv","data/storedata_total.csv")

پروسیسنگ اسکرپٹ اور پروسیسنگ مرحلہ کی وضاحت کریں۔

اس مرحلے میں، آپ فیچر انجینئرنگ، ایک ہاٹ انکوڈنگ، اور ماڈل بنانے کے لیے استعمال ہونے والی ٹریننگ، توثیق، اور ٹیسٹ اسپلٹس کو درست کرنے کے لیے ایک Python اسکرپٹ تیار کرتے ہیں۔ اپنی پروسیسنگ اسکرپٹ بنانے کے لیے درج ذیل کوڈ کو چلائیں:

%%writefile pipelines/customerchurn/preprocess.py

import os
import tempfile
import numpy as np
import pandas as pd
import datetime as dt
if __name__ == "__main__":
    base_dir = "/opt/ml/processing"
    #Read Data
    df = pd.read_csv(
        f"{base_dir}/input/storedata_total.csv"
    )
    # convert created column to datetime
    df["created"] = pd.to_datetime(df["created"])
    #Convert firstorder and lastorder to datetime datatype
    df["firstorder"] = pd.to_datetime(df["firstorder"],errors='coerce')
    df["lastorder"] = pd.to_datetime(df["lastorder"],errors='coerce')
    #Drop Rows with Null Values
    df = df.dropna()
    #Create column which gives the days between the last order and the first order
    df['first_last_days_diff'] = (df['lastorder'] - df['firstorder']).dt.days
    #Create column which gives the days between the customer record was created and the first order
    df['created_first_days_diff'] = (df['created'] - df['firstorder']).dt.days
    #Drop columns
    df.drop(['custid', 'created','firstorder','lastorder'], axis=1, inplace=True)
    #Apply one hot encoding on favday and city columns
    df = pd.get_dummies(df, prefix=['favday', 'city'], columns=['favday', 'city'])
    # Split into train, validation and test datasets
    y = df.pop("retained")
    X_pre = df
    y_pre = y.to_numpy().reshape(len(y), 1)
    X = np.concatenate((y_pre, X_pre), axis=1)
    np.random.shuffle(X)
    # Split in Train, Test and Validation Datasets
    train, validation, test = np.split(X, [int(.7*len(X)), int(.85*len(X))])
    train_rows = np.shape(train)[0]
    validation_rows = np.shape(validation)[0]
    test_rows = np.shape(test)[0]
    train = pd.DataFrame(train)
    test = pd.DataFrame(test)
    validation = pd.DataFrame(validation)
    # Convert the label column to integer
    train[0] = train[0].astype(int)
    test[0] = test[0].astype(int)
    validation[0] = validation[0].astype(int)
    # Save the Dataframes as csv files
    train.to_csv(f"{base_dir}/train/train.csv", header=False, index=False)
    validation.to_csv(f"{base_dir}/validation/validation.csv", header=False, index=False)
    test.to_csv(f"{base_dir}/test/test.csv", header=False, index=False)

اگلا، پروسیسر کو تیز کرنے کے لیے درج ذیل کوڈ بلاک کو چلائیں اور پروسیسنگ اسکرپٹ کو چلانے کے لیے پائپ لائنز کا مرحلہ۔ کیونکہ پروسیسنگ اسکرپٹ پانڈوں میں لکھا جاتا ہے، آپ استعمال کرتے ہیں a SKLearnProcessor. پائپ لائنز ProcessingStep فنکشن مندرجہ ذیل دلائل لیتا ہے: پروسیسر، خام ڈیٹاسیٹس کے لیے ان پٹ S3 مقامات، اور پروسیس شدہ ڈیٹاسیٹس کو بچانے کے لیے آؤٹ پٹ S3 مقامات۔

# Upload processing script to S3
s3_client.Bucket(default_bucket).upload_file("pipelines/customerchurn/preprocess.py","input/code/preprocess.py")

# Define Processing Step for Feature Engineering
from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.processing import ProcessingInput, ProcessingOutput
from sagemaker.workflow.steps import ProcessingStep

framework_version = "1.0-1"sklearn_processor = SKLearnProcessor(
    framework_version=framework_version,
    instance_type="ml.m5.xlarge",
    instance_count=processing_instance_count,
    base_job_name="sklearn-churn-process",
    role=role,
    sagemaker_session=pipeline_session,
)
processor_args = sklearn_processor.run(
    inputs=[
      ProcessingInput(source=input_data, destination="/opt/ml/processing/input"),  
    ],
    outputs=[
        ProcessingOutput(output_name="train", source="/opt/ml/processing/train",
                         destination=f"s3://{default_bucket}/output/train" ),
        ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation",
                        destination=f"s3://{default_bucket}/output/validation"),
        ProcessingOutput(output_name="test", source="/opt/ml/processing/test",
                        destination=f"s3://{default_bucket}/output/test")
    ],
    code=f"s3://{default_bucket}/input/code/preprocess.py",
)
step_process = ProcessingStep(name="ChurnModelProcess", step_args=processor_args)

تربیتی قدم کی وضاحت کریں۔

SageMaker XGBoost تخمینہ کار اور پائپ لائنز کا استعمال کرتے ہوئے ماڈل ٹریننگ ترتیب دیں۔ TrainingStep فنکشن:

from sagemaker.estimator import Estimator
from sagemaker.inputs import TrainingInput

model_path = f"s3://{default_bucket}/output"
image_uri = sagemaker.image_uris.retrieve(
    framework="xgboost",
    region=region,
    version="1.0-1",
    py_version="py3",
    instance_type="ml.m5.xlarge",
)
xgb_train = Estimator(
    image_uri=image_uri,
    instance_type=training_instance_type,
    instance_count=1,
    output_path=model_path,
    role=role,
    sagemaker_session=pipeline_session,
)
xgb_train.set_hyperparameters(
    objective="reg:linear",
    num_round=50,
    max_depth=5,
    eta=0.2,
    gamma=4,
    min_child_weight=6,
    subsample=0.7,
)

train_args = xgb_train.fit(
    inputs={
            "train": TrainingInput(
                s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
                    "train"
                ].S3Output.S3Uri,
                content_type="text/csv",
            ),
            "validation": TrainingInput(
                s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
                    "validation"
                ].S3Output.S3Uri,
                content_type="text/csv",
            ),
        },
)
from sagemaker.inputs import TrainingInput
from sagemaker.workflow.steps import TrainingStep
step_train = TrainingStep(
    name="ChurnModelTrain",
    step_args=train_args,
    )

تشخیصی اسکرپٹ اور ماڈل کی تشخیص کے مرحلے کی وضاحت کریں۔

ایک بار تربیت حاصل کرنے کے بعد ماڈل کا جائزہ لینے کے لیے درج ذیل کوڈ بلاک کو چلائیں۔ یہ اسکرپٹ یہ جانچنے کے لیے منطق کو سمیٹتا ہے کہ آیا AUC سکور مخصوص حد کو پورا کرتا ہے۔

%%writefile pipelines/customerchurn/evaluate.py

import json
import pathlib
import pickle
import tarfile
import joblib
import numpy as np
import pandas as pd
import xgboost
import datetime as dt
from sklearn.metrics import roc_curve,auc
if __name__ == "__main__":   
    #Read Model Tar File
    model_path = f"/opt/ml/processing/model/model.tar.gz"
    with tarfile.open(model_path) as tar:
        tar.extractall(path=".")
    model = pickle.load(open("xgboost-model", "rb"))
    #Read Test Data using which we evaluate the model
    test_path = "/opt/ml/processing/test/test.csv"
    df = pd.read_csv(test_path, header=None)
    y_test = df.iloc[:, 0].to_numpy()
    df.drop(df.columns[0], axis=1, inplace=True)
    X_test = xgboost.DMatrix(df.values)
    #Run Predictions
    predictions = model.predict(X_test)
    #Evaluate Predictions
    fpr, tpr, thresholds = roc_curve(y_test, predictions)
    auc_score = auc(fpr, tpr)
    report_dict = {
        "classification_metrics": {
            "auc_score": {
                "value": auc_score,
            },
        },
    }
    #Save Evaluation Report
    output_dir = "/opt/ml/processing/evaluation"
    pathlib.Path(output_dir).mkdir(parents=True, exist_ok=True)
    evaluation_path = f"{output_dir}/evaluation.json"
    with open(evaluation_path, "w") as f:
        f.write(json.dumps(report_dict))

اگلا، پروسیسر کو تیز کرنے کے لیے درج ذیل کوڈ بلاک کو چلائیں اور تشخیص اسکرپٹ کو چلانے کے لیے پائپ لائنز کا مرحلہ۔ چونکہ تشخیصی اسکرپٹ XGBoost پیکیج کا استعمال کرتا ہے، آپ استعمال کرتے ہیں a ScriptProcessor XGBoost امیج کے ساتھ۔ پائپ لائنز ProcessingStep فنکشن مندرجہ ذیل دلائل لیتا ہے: پروسیسر، خام ڈیٹاسیٹس کے لیے ان پٹ S3 مقامات، اور پروسیس شدہ ڈیٹاسیٹس کو بچانے کے لیے آؤٹ پٹ S3 مقامات۔

#Upload the evaluation script to S3
s3_client.Bucket(default_bucket).upload_file("pipelines/customerchurn/evaluate.py","input/code/evaluate.py")
from sagemaker.processing import ScriptProcessor
# define model evaluation step to evaluate the trained model
script_eval = ScriptProcessor(
    image_uri=image_uri,
    command=["python3"],
    instance_type=processing_instance_type,
    instance_count=1,
    base_job_name="script-churn-eval",
    role=role,
    sagemaker_session=pipeline_session,
)
eval_args = script_eval.run(
    inputs=[
        ProcessingInput(
            source=step_train.properties.ModelArtifacts.S3ModelArtifacts,
            destination="/opt/ml/processing/model",
        ),
        ProcessingInput(
            source=step_process.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri,
            destination="/opt/ml/processing/test",
        ),
    ],
    outputs=[
            ProcessingOutput(output_name="evaluation", source="/opt/ml/processing/evaluation",
                             destination=f"s3://{default_bucket}/output/evaluation"),
        ],
    code=f"s3://{default_bucket}/input/code/evaluate.py",
)
from sagemaker.workflow.properties import PropertyFile
evaluation_report = PropertyFile(
    name="ChurnEvaluationReport", output_name="evaluation", path="evaluation.json"
)
step_eval = ProcessingStep(
    name="ChurnEvalModel",
    step_args=eval_args,
    property_files=[evaluation_report],
)

تخلیق ماڈل مرحلہ کی وضاحت کریں۔

پائپ لائنز ماڈل سٹیپ کا استعمال کرتے ہوئے سیج میکر ماڈل بنانے کے لیے درج ذیل کوڈ بلاک کو چلائیں۔ یہ مرحلہ تربیتی مرحلے کے آؤٹ پٹ کو استعمال کرتا ہے تاکہ تعیناتی کے لیے ماڈل کو پیک کیا جا سکے۔ نوٹ کریں کہ مثال کی قسم کی دلیل کی قدر پائپ لائنز پیرامیٹر کا استعمال کرتے ہوئے پاس کی جاتی ہے جس کی آپ نے پوسٹ میں پہلے وضاحت کی تھی۔

from sagemaker import Model
from sagemaker.inputs import CreateModelInput
from sagemaker.workflow.model_step import ModelStep
# step to create model 
model = Model(
    image_uri=image_uri,        
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    sagemaker_session=pipeline_session,
    role=role,
)
step_create_model = ModelStep(
    name="ChurnCreateModel",
    step_args=model.create(instance_type="ml.m5.large", accelerator_type="ml.eia1.medium"),
)

بیچ کی تبدیلی کے مرحلے کی وضاحت کریں۔

پہلے مرحلے میں بنائے گئے بیچ ان پٹ کے ساتھ تربیت یافتہ ماڈل کا استعمال کرتے ہوئے بیچ ٹرانسفارمیشن کو چلانے کے لیے درج ذیل کوڈ بلاک کو چلائیں:

from sagemaker.transformer import Transformer
from sagemaker.inputs import TransformInput
from sagemaker.workflow.steps import TransformStep

transformer = Transformer(
    model_name=step_create_model.properties.ModelName,
    instance_type="ml.m5.xlarge",
    instance_count=1,
    output_path=f"s3://{default_bucket}/ChurnTransform",
    sagemaker_session=pipeline_session
)
                                 
step_transform = TransformStep(
    name="ChurnTransform", 
    step_args=transformer.transform(
                    data=batch_data,
                    content_type="text/csv"
                 )
)

رجسٹر ماڈل مرحلہ کی وضاحت کریں۔

درج ذیل کوڈ ماڈل کو SageMaker ماڈل رجسٹری میں پائپ لائنز ماڈل سٹیپ کا استعمال کرتے ہوئے رجسٹر کرتا ہے۔

model = Model(
    image_uri=image_uri,
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    sagemaker_session=pipeline_session,
    role=role,
)
from sagemaker.model_metrics import MetricsSource, ModelMetrics

model_metrics = ModelMetrics(
    model_statistics=MetricsSource(
        s3_uri="{}/evaluation.json".format(
            step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"]
        ),
        content_type="application/json",
    )
)
register_args = model.register(
    content_types=["text/csv"],
    response_types=["text/csv"],
    inference_instances=["ml.t2.medium", "ml.m5.xlarge"],
    transform_instances=["ml.m5.xlarge"],
    model_package_group_name=model_package_group_name,
    approval_status=model_approval_status,
    model_metrics=model_metrics,
)
step_register = ModelStep(name="ChurnRegisterModel", step_args=register_args)

پائپ لائن کو روکنے کے لیے ناکام قدم کی وضاحت کریں۔

اگر AUC سکور طے شدہ حد پر پورا نہیں اترتا ہے تو درج ذیل کوڈ پائپ لائنز کے ناکام ہونے کے مرحلے کی وضاحت کرتا ہے جس میں پائپ لائن کو خرابی کے پیغام کے ساتھ چلایا جاتا ہے:

from sagemaker.workflow.fail_step import FailStep
from sagemaker.workflow.functions import Join
step_fail = FailStep(
    name="ChurnAUCScoreFail",
    error_message=Join(on=" ", values=["Execution failed due to AUC Score >", auc_score_threshold]),
    )

AUC سکور کو چیک کرنے کے لیے ایک شرط کے مرحلے کی وضاحت کریں۔

مندرجہ ذیل کوڈ AUC سکور کو چیک کرنے اور مشروط طور پر ایک ماڈل بنانے اور بیچ ٹرانسفارمیشن چلانے اور ماڈل رجسٹری میں ایک ماڈل کو رجسٹر کرنے، یا پائپ لائن کو ناکام حالت میں چلانے کو روکنے کے لیے شرط کے مرحلے کی وضاحت کرتا ہے:

from sagemaker.workflow.conditions import ConditionGreaterThan
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.functions import JsonGet
cond_lte = ConditionGreaterThan(
    left=JsonGet(
        step_name=step_eval.name,
        property_file=evaluation_report,
        json_path="classification_metrics.auc_score.value",
    ),
    right=auc_score_threshold,
)
step_cond = ConditionStep(
    name="CheckAUCScoreChurnEvaluation",
    conditions=[cond_lte],
    if_steps=[step_register, step_create_model, step_transform],
    else_steps=[step_fail],
)

پائپ لائن بنائیں اور چلائیں۔

اجزاء کے تمام مراحل کی وضاحت کرنے کے بعد، آپ انہیں پائپ لائن آبجیکٹ میں جمع کر سکتے ہیں۔ آپ کو پائپ لائن کی ترتیب بتانے کی ضرورت نہیں ہے کیونکہ پائپ لائنز خود بخود مراحل کے درمیان انحصار کی بنیاد پر آرڈر کی ترتیب کا اندازہ لگاتی ہیں۔

import json
from sagemaker.workflow.pipeline import Pipeline

pipeline = Pipeline(
    name=pipeline_name,
    parameters=[
        processing_instance_count,
        processing_instance_type,
        training_instance_type,
        model_approval_status,
        input_data,
        batch_data,
        auc_score_threshold,
    ],
    steps=[step_process, step_train, step_eval, step_cond],
) 
definition = json.loads(pipeline.definition())
print(definition)

اپنی نوٹ بک کے سیل میں درج ذیل کوڈ کو چلائیں۔ اگر پائپ لائن پہلے سے موجود ہے، تو کوڈ پائپ لائن کو اپ ڈیٹ کرتا ہے۔ اگر پائپ لائن موجود نہیں ہے، تو یہ ایک نئی بناتی ہے۔

pipeline.start()
# Create a new or update existing Pipeline
pipeline.upsert(role_arn=sagemaker_role)
# start Pipeline execution

نتیجہ

اس پوسٹ میں، ہم نے کچھ نئی خصوصیات متعارف کرائیں جو اب پائپ لائنز کے ساتھ دیگر بلٹ ان SageMaker خصوصیات اور XGBoost الگورتھم کے ساتھ دستیاب ہیں تاکہ منتھن کی پیشن گوئی کے لیے ایک ماڈل تیار، اعادہ، اور تعینات کیا جا سکے۔ حل کو اضافی ڈیٹا ذرائع کے ساتھ بڑھایا جا سکتا ہے۔

اپنے ایم ایل ورک فلو کو نافذ کرنے کے لیے۔ پائپ لائنز ورک فلو میں دستیاب مراحل کے بارے میں مزید تفصیلات کے لیے، رجوع کریں۔ ایمیزون سیج میکر ماڈل بلڈنگ پائپ لائن اور سیج میکر ورک فلوز. AWS سیج میکر کی مثالیں۔ گٹ ہب ریپو میں پائپ لائنز کا استعمال کرتے ہوئے مختلف استعمال کے معاملات کے بارے میں مزید مثالیں ہیں۔

مصنفین کے بارے میں

جیری پینگ AWS SageMaker کے ساتھ سافٹ ویئر ڈویلپمنٹ انجینئر ہے۔ وہ پیداوار میں تربیت سے لے کر ماڈل کی نگرانی تک بڑے پیمانے پر MLOps سسٹم کی تعمیر پر توجہ مرکوز کرتا ہے۔ وہ MLOps کے تصور کو وسیع تر سامعین تک پہنچانے کا بھی پرجوش ہے۔

ڈیوین کیو AWS میں سافٹ ویئر ڈویلپمنٹ انجینئر ہے۔ وہ فی الحال سیج میکر پائپ لائنز کی ترقی اور بہتری پر توجہ مرکوز کر رہی ہے۔ کام سے باہر، وہ سیلو کی مشق سے لطف اندوز ہوتی ہے۔

گایتری گھاناکوٹا AWS پروفیشنل سروسز کے ساتھ ایک سینئر مشین لرننگ انجینئر ہے۔ وہ مختلف ڈومینز میں AI/ML حل تیار کرنے، تعینات کرنے اور اس کی وضاحت کرنے کا شوق رکھتی ہے۔ اس کردار سے پہلے، اس نے ایک ڈیٹا سائنسدان اور ایم ایل انجینئر کے طور پر مالیاتی اور خوردہ جگہ میں اعلی عالمی فرموں کے ساتھ متعدد اقدامات کی قیادت کی۔ اس نے یونیورسٹی آف کولوراڈو، بولڈر سے ڈیٹا سائنس میں کمپیوٹر سائنس میں ماسٹر کی ڈگری حاصل کی ہے۔

روپندر گریوال AWS کے ساتھ ایک Sr Ai/ML ماہر حل آرکیٹیکٹ ہے۔ وہ فی الحال SageMaker پر ماڈلز اور MLOps کی خدمت پر توجہ مرکوز کرتا ہے۔ اس کردار سے پہلے وہ مشین لرننگ انجینئر بلڈنگ اور ہوسٹنگ ماڈل کے طور پر کام کر چکے ہیں۔ کام کے علاوہ وہ ٹینس کھیلنا اور پہاڑی پگڈنڈیوں پر بائیک چلانا پسند کرتا ہے۔

رے لی AWS پروفیشنل سروسز کے ساتھ ایک سینئر ڈیٹا سائنٹسٹ ہے۔ اس کی خاصیت مختلف سائز کے صارفین کے لیے AI/ML سلوشنز بنانے اور چلانے پر مرکوز ہے، جس میں اسٹارٹ اپس سے لے کر انٹرپرائز آرگنائزیشنز شامل ہیں۔ کام سے باہر، رے فٹنس اور سفر سے لطف اندوز ہوتا ہے۔