+
+ ***Now that the model is registered, you can get access to the
+ registered model manually on the sagemaker studio model registry
+ console, or programmatically in the next notebook, approve it and
+ run the second portion of this solution: Batch Transform Step***
+
+2. Next Choose
+ [`1_batch_transform_pipeline.ipynb`](https://github.com/aws-samples/mlops-pipeline-prestodb/blob/main/1_batch_transform_pipeline.ipynb).
+ When the notebook is open, on the Run menu, choose **Run All Cells**
+ to run the code in this notebook. This notebook will run a batch
+ transform step using the model trained in the previous notebook. It
+ does so by running the following steps:
+
+ - **Extract the latest approved model from the SageMaker model
+ registry**: In this step, we first define pipeline input
+ parameters that are used for the ***EC2 instance types to use
+ for processing and training steps***. These parameters can be
+ configured on the [config.yml](./config.yml) file.
+
+ ``` python
+ # What instance type to use for processing.
+ processing_instance_type = ParameterString(
+ name="ProcessingInstanceType", default_value=config['data_processing_step']['processing_instance_type']
+ )
+
+
+ # Create SKlearn processor object,
+ # The object contains information about what instance type to use, the IAM role to use etc.
+ # A managed processor comes with a preconfigured container, so only specifying version is required.
+
+ est_cls = sagemaker.sklearn.estimator.SKLearn
+
+ sklearn_processor = FrameworkProcessor(
+ estimator_cls=est_cls,
+ framework_version=config['training_step']['sklearn_framework_version'],
+ role=role,
+ instance_type=processing_instance_type,
+ instance_count=config['data_processing_step']['instance_count'],
+ tags=config['data_processing_step']['tags'],
+ sagemaker_session=pipeline_session,
+ base_job_name=config['pipeline']['base_job_name'], )
+ ```
+
+ Now, we use an image URI that we create to extract the latest model
+ that was approved from the model registry, and set the
+ `ModelApprovalStatus` to `Approved`:
+
+ ``` python
+ # list all model packages and select the latest one
+ model_packages = []
+
+ for p in sm.get_paginator('list_model_packages').paginate(
+ ModelPackageGroupName=config['register_model_step']['model_group'],
+ SortBy="CreationTime",
+ SortOrder="Descending",
+ ):
+ model_packages.extend(p["ModelPackageSummaryList"])
+
+ if len(model_packages) == 0:
+ raise Exception(f"No model package is found for {config['register_model_step']['model_group']} model package group")
+
+ ## print the latest model, approve it
+ latest_model_package_arn = model_packages[0]["ModelPackageArn"]
+
+ ## updating the latest model package to approved status to use it for batch inference
+ model_package_update_response = sm.update_model_package(
+ ModelPackageArn=latest_model_package_arn,
+ ModelApprovalStatus="Approved",
+ )
+ ```
+
+ Now we have extracted the latest model from the SageMaker Model
+ Registry, and programmatically approved it. You can also approve the
+ model manually on the [SageMaker Model
+ Registry](https://docs.aws.amazon.com/sagemaker/latest/dg/model-registry.html)
+ page on SageMaker Studio as given in the screenshot below.
+ 
+
+ - **Read raw data for inference from PrestoDB and store in an
+ Amazon S3 bucket**: Once the latest model is approved, we get
+ the latest batch data from presto and use that for our batch
+ transform step. In this step, we use another [batch preprocess
+ script](https://github.com/aws-samples/mlops-pipeline-prestodb/blob/main/code/presto_preprocess_for_batch_inference.py)
+ that is dedicated to reading and fetching data from presto and
+ saving in a batch directory within our S3 bucket.
+
+ ``` python
+ ## represents the output processing for the batch pre processing step
+ batch_output=[
+ ProcessingOutput(
+ output_name="batch",
+ source="/opt/ml/processing/batch",
+ destination=Join(
+ on="/",
+ values=[
+ "s3://{}".format(bucket),
+ prefix,
+ ExecutionVariables.PIPELINE_EXECUTION_ID,
+ "batch",
+ ],
+ ),
+ ),
+ ]
+
+
+ # Use the sklearn_processor's run method and configure the batch preprocessing step
+ step_args = sklearn_processor.run(
+ # here, we add in a `code` or an entry point that uses the data preprocess script for collecting data in a batch and storing it in S3
+ code=config['scripts']['batch_transform_get_data'],
+ source_dir=config['scripts']['source_dir'],
+ outputs=batch_output,
+ arguments=[
+ "--host", host_parameter,
+ "--port", port_parameter,
+ "--presto_credentials_key", presto_parameter,
+ "--region", region_parameter,
+ "--presto_catalog", presto_catalog_parameter,
+ "--presto_schema", presto_schema_parameter,
+ ],
+ )
+
+ # declare the batch step that is called later in pipeline execution
+ batch_data_prep = ProcessingStep(
+ name=config['data_processing_step']['step_name'],
+ step_args=step_args,
+ )
+ ```
+
+ Once the batch data preparation step is complete, we declare a model
+ with the image uri and refer to the
+ [‘inference.py’](https://github.com/aws-samples/mlops-pipeline-prestodb/blob/main/code/inference.py)
+ script that grabs information on features to use while making
+ predictions. Using this, we will create the model which
+ automatically trigger the training and the preprocess data step Run
+ the transformer step on the created model.
+
+ ``` python
+ # create the model image based on the model data and refer to the inference script as an entry point for batch inference
+ model = Model(
+ image_uri=image_uri,
+ entry_point=config['scripts']['batch_inference'],
+ model_data=model_data_url,
+ sagemaker_session=pipeline_session,
+ role=role,
+ )
+ ```
+
+ - **Create a batch transform step to provide inference on the
+ data. The inference results are also stored in S3**: Now that a
+ model instance is defined, create a Transformer instance with
+ the appropriate model type, compute instance type, and desired
+ output S3 URI. Specifically, pass in the ModelName from the
+ CreateModelStep, step_create_model properties. The
+ CreateModelStep properties attribute matches the object model of
+ the DescribeModel response object.
+
+ ``` python
+ transformer = Transformer(
+ model_name=step_create_model.properties.ModelName,
+ instance_type=config['transform_step']['instance_type'],
+ instance_count=config['transform_step']['instance_count'],
+ strategy="MultiRecord",
+ accept="text/csv",
+ assemble_with="Line",
+ output_path=f"s3://{bucket}",
+ tags = config['transform_step']['tags'],
+ env={
+ 'START_TIME_UTC': st.strftime('%Y-%m-%d %H:%M:%S'),
+ 'END_TIME_UTC': et.strftime('%Y-%m-%d %H:%M:%S'),
+ })
+ ```
+
+ Now that our transformer object is created, we pass the transformer
+ input (that contains the batch data from our batch pre process step)
+ into the step declaration:
+
+ ``` python
+ step_transform = TransformStep(
+ name=config['transform_step']['step_name'], transformer=transformer, inputs=transform_input,
+ )
+
+ batch_transform_pipeline = Pipeline(
+ name=pipeline_name,
+ parameters=[processing_instance_type,
+ host_parameter,
+ presto_parameter,
+ region_parameter,
+ port_parameter,
+ target_parameter,
+ feature_parameter,
+ presto_catalog_parameter,
+ presto_schema_parameter,],
+ steps=[
+ batch_data_prep,
+ step_create_model,
+ step_transform,
+ ],
+ )
+
+ # start the pipeline execution:
+ response = sagemaker_client.start_pipeline_execution(
+ PipelineName=batch_transform_pipeline.name
+ )
+
+ while True:
+ resp = client.describe_pipeline_execution(
+ PipelineExecutionArn=response['PipelineExecutionArn']
+ )
+ status = resp['PipelineExecutionStatus']
+ ```
+
+ **At the end of the batch transform pipeline, your pipeline
+ structure on Amazon SageMaker Pipelines should look like this:**
+ 
+
+3. Lastly, Choose
+ [`2_realtime_inference.ipynb`](https://github.com/aws-samples/mlops-pipeline-prestodb/blob/main/2_realtime_inference.ipynb).
+ When the notebook is open, on the Run menu, choose **Run All Cells**
+ to run the code in this notebook. This notebook extracts the latest
+ approved model from the model registry and deploys it as a SageMaker
+ endpoint for real time inference. It does so by running the
+ following steps:
+
+ - **Extract the latest approved model from the SageMaker model
+ registry**: To deploy a real time SageMaker endpoint, we first
+ will fetch the `image uri` to use and extract the latest
+ approved model the same way we did in the prior batch transfrom
+ notebook. Once you have extracted the latest approved model, use
+ a container list with the specific `inference.py` file to create
+ the model and run inferences against. This model creation and
+ endpoint deployment is specific to the [Scikit-learn
+ model](https://sagemaker.readthedocs.io/en/stable/frameworks/sklearn/sagemaker.sklearn.html)
+ configuration and will change based on your use case.
+
+ ``` python
+ container_list = [{
+ 'Image': image_uri,
+ 'ModelDataUrl': model_data_url,
+ 'Environment': {
+ 'SAGEMAKER_PROGRAM': 'inference.py',
+ 'SAGEMAKER_SUBMIT_DIRECTORY': compressed_inference_script_uri,
+ }
+ }]
+
+ ## create the model object and call deploy on it
+ create_model_response = sm.create_model(
+ ModelName = model_name,
+ ExecutionRoleArn = role,
+ Containers=container_list
+ )
+ ```
+
+ In this code, we use the inference.py file specific to the Scikit
+ Learn model. We then create our endpoint configuration, setting our
+ `ManagedInstanceScaling` to `ENABLED` with our desired
+ `MaxInstanceCount` and `MinInstanceCount` for automatic scaling.
+
+ ``` python
+ create_endpoint_config_response = sm.create_endpoint_config(
+ EndpointConfigName = endpoint_config_name,
+ ProductionVariants=[{
+ 'InstanceType': instance_type,
+ ## have max instance count configured here
+ 'InitialInstanceCount': min_instances,
+ 'InitialVariantWeight': 1,
+ 'ModelName': model_name,
+ 'VariantName': 'AllTraffic',
+ ## change your managed instance configuration here
+ "ManagedInstanceScaling":{
+ "MaxInstanceCount": max_instances,
+ "MinInstanceCount": min_instances,
+ "Status": "ENABLED",}
+ }])
+ ```
+
+ - **Runs inferences for testing the real time deployed endpoint**:
+ Once you have extracted the latest approved model, created the
+ model from the desired image uri and configured the endpoint
+ configuration, you can then deploy it as a real time SageMaker
+ endpoint below:
+
+ ``` python
+ create_endpoint_response = sm.create_endpoint(
+ EndpointName=endpoint_name,
+ EndpointConfigName=endpoint_config_name)
+ logger.info(f"Going to deploy the real time endpoint -> {create_endpoint_response['EndpointArn']}")
+
+ # wait for endpoint to reach a terminal state (InService) using describe endpoint
+ describe_endpoint_response = sm.describe_endpoint(EndpointName=endpoint_name)
+
+ while describe_endpoint_response["EndpointStatus"] == "Creating":
+ describe_endpoint_response = sm.describe_endpoint(EndpointName=endpoint_name)
+ print(describe_endpoint_response["EndpointStatus"])
+ time.sleep(15)
+ ```
+
+ Now run inference against the data extracted from prestoDB:
+
+ ``` python
+ body_str = "total_extended_price,avg_discount,total_quantity\n1,2,3\n66.77,12,2"
+
+ response = smr.invoke_endpoint(
+ EndpointName=endpoint_name,
+ Body=body_str.encode('utf-8') ,
+ ContentType='text/csv',
+ )
+
+ response_str = response["Body"].read().decode()
+ response_str
+ ```
+
+ We have now seen an end to end process of our solution, from
+ fetching data by connecting to a Presto Server on an EC2 instance,
+ followed by training, evaluating, registering the model. We then
+ approved the latest registered model from our training pipeline
+ solution and ran batch inference against batch data stored in S3. We
+ finally deployed the latest approved model as a real time SageMaker
+ endpoint to run inferences against. Take a look at the results
+ below.
+
+## Results
+
+Here is a compilation of some queries and responses generated by our
+implementation from the real time endpoint deployment stage:
+
+| Query | +Answer | +
|---|---|
| total_extended_price,avg_discount,total_quantity,2,3,12,2 | +– response – | +