Amazon SageMaker Pipelines এবং Amazon SageMaker 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 এবং আপনার জন্য অন্যান্য সেজমেকার সত্তা এবং সংস্থানগুলির সাথে ইন্টারঅ্যাক্ট করার জন্য সুবিধাজনক পদ্ধতি প্রয়োগ করে, যেমন প্রশিক্ষণের কাজ, এন্ডপয়েন্ট এবং ইনপুট ডেটাসেট সংরক্ষিত আমাজন সিম্পল স্টোরেজ সার্ভিস (Amazon 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 প্রকার উল্লেখ্য যে ক 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)। স্টুডিও এন্ড-টু-এন্ড পাইপলাইন অভিজ্ঞতা পরিচালনা করার জন্য একটি পরিবেশ সরবরাহ করে। স্টুডিও দিয়ে, আপনি বাইপাস করতে পারেন এডাব্লুএস ম্যানেজমেন্ট কনসোল আপনার সম্পূর্ণ কর্মপ্রবাহ ব্যবস্থাপনার জন্য। স্টুডিওর মধ্যে থেকে পাইপলাইন পরিচালনার বিষয়ে আরও তথ্যের জন্য, পড়ুন সেজমেকার স্টুডিওতে সেজমেকার পাইপলাইনগুলি দেখুন, ট্র্যাক করুন এবং কার্যকর করুন৷.

নিম্নলিখিত চিত্রটি নতুন বৈশিষ্ট্যগুলি ব্যবহার করে প্রশিক্ষণ এবং অনুমান তৈরি করার বিভিন্ন পদক্ষেপ সহ ML কর্মপ্রবাহের উচ্চ-স্তরের আর্কিটেকচারকে চিত্রিত করে।

পাইপলাইনে নিম্নলিখিত পদক্ষেপগুলি অন্তর্ভুক্ত রয়েছে:

1. প্রয়োজনীয় বৈশিষ্ট্যগুলি তৈরি করতে এবং ট্রেন, বৈধতা এবং পরীক্ষার ডেটাসেটে ডেটা বিভক্ত করার জন্য ডেটা প্রিপ্রসেস করুন।
2. SageMaker XGBoost ফ্রেমওয়ার্কের সাথে একটি প্রশিক্ষণের কাজ তৈরি করুন।
3. পরীক্ষার ডেটাসেট ব্যবহার করে প্রশিক্ষিত মডেলের মূল্যায়ন করুন।
4. AUC স্কোর পূর্বনির্ধারিত থ্রেশহোল্ডের উপরে কিনা তা পরীক্ষা করুন।
   • AUC স্কোর থ্রেশহোল্ডের চেয়ে কম হলে, পাইপলাইন চালানো বন্ধ করুন এবং এটি ব্যর্থ হিসাবে চিহ্নিত করুন।
   • যদি AUC স্কোর থ্রেশহোল্ডের চেয়ে বেশি হয়, একটি SageMaker মডেল তৈরি করুন এবং SageMaker মডেল রেজিস্ট্রিতে এটি নিবন্ধন করুন।
5. পূর্ববর্তী ধাপে তৈরি মডেল ব্যবহার করে প্রদত্ত ডেটাসেটে ব্যাচ রূপান্তর প্রয়োগ করুন।

পূর্বশর্ত

এই পোস্টটি অনুসরণ করার জন্য, আপনার একটি AWS অ্যাকাউন্ট প্রয়োজন স্টুডিও ডোমেইন.

পাইপলাইনগুলি সরাসরি সেজমেকার সত্তা এবং সংস্থানগুলির সাথে একত্রিত করা হয়েছে, তাই আপনাকে অন্য কোনও AWS পরিষেবাগুলির সাথে ইন্টারঅ্যাক্ট করতে হবে না৷ আপনাকে কোনো সংস্থান পরিচালনা করতে হবে না কারণ এটি একটি সম্পূর্ণরূপে পরিচালিত পরিষেবা, যার অর্থ এটি আপনার জন্য সংস্থান তৈরি এবং পরিচালনা করে৷ স্টুডিওর ইন্টিগ্রেটেড কম্পোনেন্ট সহ বিভিন্ন সেজমেকার উপাদানগুলির বিষয়ে আরও তথ্যের জন্য দেখুন SageMaker পণ্য পাতা.

শুরু করার আগে, নিম্নলিখিত কোড স্নিপেট ব্যবহার করে স্টুডিও নোটবুকের মধ্যে 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

এমএল কর্মপ্রবাহ

এই পোস্টের জন্য, আপনি নিম্নলিখিত উপাদান ব্যবহার করুন:

• ডেটা প্রস্তুতি
  • সেজমেকার প্রসেসিং - সেজমেকার প্রসেসিং একটি সম্পূর্ণরূপে পরিচালিত পরিষেবা যা আপনাকে ML কাজের চাপের জন্য কাস্টম ডেটা ট্রান্সফরমেশন এবং ফিচার ইঞ্জিনিয়ারিং চালানোর অনুমতি দেয়।
• মডেল ভবন
• মডেল প্রশিক্ষণ এবং মূল্যায়ন
  • এক-ক্লিক প্রশিক্ষণ - সেজমেকার প্রশিক্ষণ বৈশিষ্ট্য বিতরণ করেছে। SageMaker ডেটা সমান্তরালতা এবং মডেল সমান্তরালতার জন্য বিতরণ করা প্রশিক্ষণ লাইব্রেরি প্রদান করে। লাইব্রেরিগুলি সেজমেকার প্রশিক্ষণ পরিবেশের জন্য অপ্টিমাইজ করা হয়েছে, আপনার বিতরণকৃত প্রশিক্ষণের কাজগুলিকে সেজমেকারের সাথে মানিয়ে নিতে এবং প্রশিক্ষণের গতি এবং থ্রুপুট উন্নত করতে সহায়তা করে৷
  • সেজমেকার পরীক্ষা-নিরীক্ষা - পরীক্ষাগুলি হল সেজমেকারের একটি ক্ষমতা যা আপনাকে আপনার ML পুনরাবৃত্তিগুলি সংগঠিত করতে, ট্র্যাক করতে, তুলনা করতে এবং মূল্যায়ন করতে দেয়৷
  • সেজমেকার ব্যাচের রূপান্তর - ব্যাচ ট্রান্সফর্ম বা অফলাইন স্কোরিং হল সেজমেকারে একটি পরিচালিত পরিষেবা যা আপনাকে আপনার এমএল মডেলগুলি ব্যবহার করে একটি বড় ডেটাসেটের পূর্বাভাস দিতে দেয়৷
• ওয়ার্কফ্লো অর্কেস্ট্রেশন

একটি সেজমেকার পাইপলাইন হল একটি JSON পাইপলাইন সংজ্ঞা দ্বারা সংজ্ঞায়িত আন্তঃসংযুক্ত পদক্ষেপগুলির একটি সিরিজ। এটি একটি নির্দেশিত অ্যাসাইক্লিক গ্রাফ (DAG) ব্যবহার করে একটি পাইপলাইন এনকোড করে। DAG পাইপলাইনের প্রতিটি ধাপের জন্য প্রয়োজনীয়তা এবং সম্পর্কের তথ্য দেয় এবং এর গঠন ধাপগুলির মধ্যে ডেটা নির্ভরতা দ্বারা নির্ধারিত হয়। এই নির্ভরতাগুলি তৈরি হয় যখন একটি ধাপের আউটপুটের বৈশিষ্ট্যগুলি অন্য ধাপে ইনপুট হিসাবে পাস করা হয়।

নিচের চিত্রটি সেজমেকার পাইপলাইনের বিভিন্ন ধাপকে চিত্রিত করে (একটি মন্থন পূর্বাভাস ব্যবহারের ক্ষেত্রে) যেখানে ধাপগুলির মধ্যে সংযোগগুলি সেজমেকার দ্বারা অনুমান করা হয় ধাপের সংজ্ঞা দ্বারা সংজ্ঞায়িত ইনপুট এবং আউটপুটগুলির উপর ভিত্তি করে৷

পরবর্তী বিভাগগুলি পাইপলাইনের প্রতিটি ধাপ তৈরি করে এবং একবার তৈরি হয়ে গেলে পুরো পাইপলাইনটি চালানোর মধ্য দিয়ে চলে।

প্রকল্পের কাঠামো

প্রকল্পের কাঠামো দিয়ে শুরু করা যাক:

• /sm-পাইপলাইন-এন্ড-টু-এন্ড-উদাহরণ - প্রকল্পের নাম
  • / তথ্য - ডেটাসেট
  • /পাইপলাইন - পাইপলাইনের উপাদানগুলির জন্য কোড ফাইল
    • /গ্রাহক মন্থন
      • preprocess.py
      • evaluate.py
  • sagemaker-pipelines-project.ipynb - পাইপলাইনের নতুন বৈশিষ্ট্যগুলি ব্যবহার করে মডেলিং কর্মপ্রবাহের মধ্য দিয়ে চলা একটি নোটবুক৷

ডেটাসেট ডাউনলোড করুন

এই পোস্টটি অনুসরণ করতে, আপনাকে ডাউনলোড করে সংরক্ষণ করতে হবে নমুনা ডেটাসেট প্রকল্প হোম ডিরেক্টরির মধ্যে ডেটা ফোল্ডারের অধীনে, যা ফাইলটিকে সংরক্ষণ করে আমাজন ইলাস্টিক ফাইল সিস্টেম (Amazon EFS) স্টুডিও পরিবেশের মধ্যে।

পাইপলাইনের উপাদানগুলি তৈরি করুন

এখন আপনি পাইপলাইনের উপাদানগুলি তৈরি করতে প্রস্তুত৷

বিবৃতি আমদানি করুন এবং প্যারামিটার এবং ধ্রুবক ঘোষণা করুন

নামে একটি স্টুডিও নোটবুক তৈরি করুন sagemaker-pipelines-project.ipynb প্রোজেক্ট হোম ডিরেক্টরির মধ্যে। একটি কক্ষে নিম্নলিখিত কোড ব্লক লিখুন, এবং সেজমেকার এবং S3 ক্লায়েন্ট অবজেক্ট সেট আপ করতে সেলটি চালান, তৈরি করুন PipelineSession, এবং সেজমেকার সেশনের সাথে আসা ডিফল্ট বালতি ব্যবহার করে S3 বালতি অবস্থান সেট আপ করুন:

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 বালতিতে ডেটা আপলোড করুন

Amazon 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")

একটি প্রক্রিয়াকরণ স্ক্রিপ্ট এবং প্রক্রিয়াকরণ পদক্ষেপ সংজ্ঞায়িত করুন

এই ধাপে, আপনি বৈশিষ্ট্য ইঞ্জিনিয়ারিং, একটি হট এনকোডিং করার জন্য একটি পাইথন স্ক্রিপ্ট প্রস্তুত করেন এবং মডেল বিল্ডিংয়ের জন্য ব্যবহার করা প্রশিক্ষণ, বৈধতা এবং পরীক্ষা বিভাজনগুলিকে কিউরেট করেন। আপনার প্রসেসিং স্ক্রিপ্ট তৈরি করতে নিম্নলিখিত কোডটি চালান:

%%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)

এর পরে, প্রসেসরকে ইনস্ট্যান্টিয়েট করতে নিম্নলিখিত কোড ব্লকটি চালান এবং প্রক্রিয়াকরণ স্ক্রিপ্ট চালানোর জন্য পাইপলাইন ধাপটি চালান। কারণ প্রক্রিয়াকরণ স্ক্রিপ্টটি পান্ডাসে লেখা হয়, আপনি একটি ব্যবহার করেন 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 প্যাকেজ ব্যবহার করে, আপনি একটি ব্যবহার করেন 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"
                 )
)

একটি রেজিস্টার মডেল ধাপ সংজ্ঞায়িত করুন

নিম্নলিখিত কোডটি পাইপলাইন মডেল ধাপ ব্যবহার করে সেজমেকার মডেল রেজিস্ট্রির মধ্যে মডেলটিকে নিবন্ধন করে:

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

উপসংহার

এই পোস্টে, আমরা মন্থন পূর্বাভাসের জন্য একটি মডেল বিকাশ, পুনরাবৃত্তি এবং স্থাপন করার জন্য অন্যান্য অন্তর্নির্মিত সেজমেকার বৈশিষ্ট্য এবং XGBoost অ্যালগরিদম সহ পাইপলাইনগুলির সাথে এখন উপলব্ধ কিছু নতুন বৈশিষ্ট্যগুলি উপস্থাপন করেছি। সমাধানটি অতিরিক্ত ডেটা উত্স দিয়ে প্রসারিত করা যেতে পারে

আপনার নিজের এমএল ওয়ার্কফ্লো বাস্তবায়ন করতে। পাইপলাইন ওয়ার্কফ্লোতে উপলব্ধ পদক্ষেপগুলি সম্পর্কে আরও বিশদ বিবরণের জন্য, পড়ুন আমাজন সেজমেকার মডেল বিল্ডিং পাইপলাইন এবং সেজমেকার ওয়ার্কফ্লোস. দ্য AWS SageMaker উদাহরণ GitHub রেপোতে পাইপলাইন ব্যবহার করে বিভিন্ন ব্যবহারের ক্ষেত্রে আরও উদাহরণ রয়েছে।

লেখক সম্পর্কে

জেরি পেং AWS SageMaker এর একজন সফটওয়্যার ডেভেলপমেন্ট ইঞ্জিনিয়ার। তিনি প্রশিক্ষণ থেকে শুরু করে উৎপাদনে মডেল মনিটরিং পর্যন্ত এন্ড-টু-এন্ড বৃহৎ-স্কেল MLOps সিস্টেম তৈরির দিকে মনোনিবেশ করেন। তিনি আরও বৃহত্তর শ্রোতাদের কাছে MLOps-এর ধারণা নিয়ে আসার বিষয়ে উত্সাহী।

ডিউয়েন কিউই AWS-এর একজন সফটওয়্যার ডেভেলপমেন্ট ইঞ্জিনিয়ার। তিনি বর্তমানে সেজমেকার পাইপলাইন উন্নয়ন এবং উন্নতিতে মনোনিবেশ করেন। কাজের বাইরে, তিনি সেলো অনুশীলন উপভোগ করেন।

গায়ত্রী ঘনাকোটা AWS পেশাদার পরিষেবা সহ একজন সিনিয়র মেশিন লার্নিং ইঞ্জিনিয়ার। তিনি বিভিন্ন ডোমেন জুড়ে AI/ML সমাধানগুলি বিকাশ, স্থাপন এবং ব্যাখ্যা করার বিষয়ে উত্সাহী। এই ভূমিকার আগে, তিনি আর্থিক এবং খুচরা জায়গায় শীর্ষ বিশ্বব্যাপী সংস্থাগুলির সাথে ডেটা বিজ্ঞানী এবং এমএল ইঞ্জিনিয়ার হিসাবে একাধিক উদ্যোগের নেতৃত্ব দিয়েছিলেন। তিনি কলোরাডো ইউনিভার্সিটি, বোল্ডার থেকে ডেটা সায়েন্সে বিশেষায়িত কম্পিউটার সায়েন্সে স্নাতকোত্তর ডিগ্রি অর্জন করেছেন।

রুপিন্দর গ্রেওয়াল AWS সহ একজন Sr Ai/ML বিশেষজ্ঞ সমাধান স্থপতি। তিনি বর্তমানে সেজমেকারে মডেল এবং এমএলওপ পরিবেশন করার দিকে মনোনিবেশ করেন। এই ভূমিকার আগে তিনি মেশিন লার্নিং ইঞ্জিনিয়ার বিল্ডিং এবং হোস্টিং মডেল হিসাবে কাজ করেছেন। কাজের বাইরে সে টেনিস খেলা এবং পাহাড়ের পথে বাইক চালানো উপভোগ করে।

রে লি AWS প্রফেশনাল সার্ভিসের সাথে একজন সিনিয়র ডেটা সায়েন্টিস্ট। তার বিশেষত্ব স্টার্টআপ থেকে শুরু করে এন্টারপ্রাইজ সংস্থা পর্যন্ত বিভিন্ন আকারের গ্রাহকদের জন্য এআই/এমএল সমাধান তৈরি এবং কার্যকর করার উপর দৃষ্টি নিবদ্ধ করে। কাজের বাইরে, রায় ফিটনেস এবং ভ্রমণ উপভোগ করেন।