<jupyter_start><jupyter_text>Scheduling SKLearn with Amazon SageMaker PipelinesIn this notebook, we will use Amazon SageMaker Pipelines to create two workflows with Scikit-Learn. We will create a pipeline that preprocess data and trains a model (we will use scikit-learn Pipeline), then we will schedule inference with SageMaker Batch Transform.  The datasetFor this example, we will create an artificial dataset with the `sklearn.datasets.make_classification()` function from the scikit-learn library. Once that's created, we will store it locally and then in S3.<jupyter_code>from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
import pandas as pd
from sagemaker import Session

session = Session()
bucket = session.default_bucket()  # Change to another bucket if running outside of SageMaker
prefix = "sklearn-pipeline/data"  # Choose your preferred prefix, but keep it consistent

# Create a random dataset for classification
X, y = make_classification(random_state=42)
data = pd.concat([pd.DataFrame(X), pd.DataFrame(y, columns=["y"])], axis=1)
data.to_csv("/tmp/data.csv", index=False)
X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=0)
pd.DataFrame(X_test).to_csv("/tmp/x_test.csv", index=False, header=False)
pd.DataFrame(y_test).to_csv("/tmp/y_test.csv", index=False, header=False)
# Upload to S3
data_path = session.upload_data(path="/tmp/data.csv", bucket=bucket, key_prefix=f"{prefix}/source")
x_test_path = session.upload_data(
    path="/tmp/x_test.csv", bucket=bucket, key_prefix=f"{prefix}/test"
)
y_test_path = session.upload_data(
    path="/tmp/y_test.csv", bucket=bucket, key_prefix=f"{prefix}/test"
)

print(data_path)<jupyter_output><empty_output><jupyter_text>SageMaker helper variables<jupyter_code>import sagemaker
import boto3

sess = sagemaker.session.Session()
bucket = sess.default_bucket()
role = sagemaker.get_execution_role()
region = boto3.Session().region_name<jupyter_output><empty_output><jupyter_text>Training pipelineLet's start by creating the preprocessing and training pipeline. In this case, we will use two pre-existing scripts, `preprocessing.py` and `training.py`, to preprocess our input data and train our model. Both scripts use the `sklearn.pipeline` library, and are expected to output a `joblib` compressed file to be re-used during inference.  Step 1 - Create the parameters of the pipelineBefore we begin to create the pipeline itself, we should think about how to parameterize it. For example, we may use different instance types for different purposes, such as CPU-based types for data processing and GPU-based or more powerful types for model training. These are all "knobs" of the pipeline that we can parameterize. Parameterizing enables custom pipeline executions and schedules without having to modify the pipeline definition.<jupyter_code>from sagemaker.workflow.parameters import (
    ParameterInteger,
    ParameterString,
)

# raw input data
input_data = ParameterString(name="InputData", default_value=data_path)

# processing step parameters
processing_instance_type = ParameterString(
    name="ProcessingInstanceType", default_value="ml.m5.xlarge"
)
processing_instance_count = ParameterInteger(name="ProcessingInstanceCount", default_value=1)

# training step parameters
training_instance_type = ParameterString(name="TrainingInstanceType", default_value="ml.c5.2xlarge")
training_instance_count = ParameterInteger(name="TrainingInstanceCount", default_value=1)

# batch inference step parameters
batch_instance_type = ParameterString(name="BatchInstanceType", default_value="ml.c5.xlarge")
batch_instance_count = ParameterInteger(name="BatchInstanceCount", default_value=1)<jupyter_output><empty_output><jupyter_text>Step 2 - Create the `SKLearnProcessor` and the `ProcessorStep`The first step in the pipeline will preprocess the data to prepare it for training. We create a `SKLearnProcessor` object similar to the one above, but now parameterized, so we can separately track and change the job configuration as needed, for example to increase the instance type size and count to accommodate a growing dataset.<jupyter_code>from sagemaker.sklearn.processing import SKLearnProcessor


framework_version = "0.23-1"

sklearn_processor = SKLearnProcessor(
    framework_version=framework_version,
    instance_type=processing_instance_type,
    instance_count=processing_instance_count,
    base_job_name="scheduled-pipelines-sklearn",
    sagemaker_session=sess,
    role=role,
)
from sagemaker.processing import ProcessingInput, ProcessingOutput
from sagemaker.workflow.steps import ProcessingStep


step_process = ProcessingStep(
    name="scheduled-pipeline-sklearn-process",
    processor=sklearn_processor,
    inputs=[
        ProcessingInput(
            source=input_data,
            destination="/opt/ml/processing/input",
            s3_data_distribution_type="ShardedByS3Key",
        ),
    ],
    outputs=[
        ProcessingOutput(output_name="train", source="/opt/ml/processing/output/train"),
        ProcessingOutput(output_name="test", source="/opt/ml/processing/output/test"),
        ProcessingOutput(output_name="pipeline", source="/opt/ml/processing/output/pipeline"),
    ],
    code="./preprocessing.py",
)<jupyter_output><empty_output><jupyter_text>Step 3 - Create the `SKLearnEstimator` and its `TrainingStep`Next, we specify a `Estimator` object, and define a `TrainingStep` to insert the training job in the pipeline with inputs from the previous SageMaker Processing step. Notice that we have used the hyperparameters from the best estimator in the tuning job we ran before.<jupyter_code>from sagemaker.sklearn.estimator import SKLearn

# Define the Estimator from SageMaker (Script Mode)
sklearn_estimator = SKLearn(
    entry_point="training.py",
    role=role,
    instance_count=training_instance_count,
    instance_type=training_instance_type,
    framework_version=framework_version,
    base_job_name="sklearn-training",
    metric_definitions=[{"Name": "model_accuracy", "Regex": "Model Accuracy: ([0-9.]+).*$"}],
    hyperparameters={"n-estimators": 100, "min-samples-leaf": 3},
)
from sagemaker.workflow.steps import TrainingStep
from sagemaker.inputs import TrainingInput

step_train = TrainingStep(
    name="scheduled-pipeline-sklearn-training",
    estimator=sklearn_estimator,
    inputs={
        "train": TrainingInput(
            s3_data=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri
        ),
        "test": TrainingInput(
            s3_data=step_process.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri
        ),
        "pipeline": TrainingInput(
            s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
                "pipeline"
            ].S3Output.S3Uri
        ),
    },
)<jupyter_output><empty_output><jupyter_text>Step 4 - Create the `SKLearnModel` abstractionAs another step, we create a SageMaker `SKLearnModel` object to wrap the model artifact, and associate it with a separate SageMaker prebuilt SKLearn inference container to potentially use later for validation and inference.<jupyter_code>from sagemaker.sklearn.model import SKLearnModel
from sagemaker.inputs import CreateModelInput
from sagemaker.workflow.steps import CreateModelStep
from sagemaker.utils import name_from_base

model = SKLearnModel(
    entry_point="training.py",
    framework_version=framework_version,
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    sagemaker_session=sess,
    role=role,
    name=name_from_base("sklearn-pipeline-model"),
)

inputs_model = CreateModelInput(instance_type=batch_instance_type)

step_create_model = CreateModelStep(
    name="scheduled-pipeline-sklearn-createmodel",
    model=model,
    inputs=inputs_model,
)<jupyter_output><empty_output><jupyter_text>Create and execute the PipelineWith all the pipeline steps now defined, we can define the pipeline itself as a `Pipeline` object comprising a series of those steps. Parallelized and conditional steps also are possible.<jupyter_code>from sagemaker.workflow.pipeline import Pipeline

pipeline_name = "sklearn-scheduled-training"

pipeline = Pipeline(
    name=pipeline_name,
    parameters=[
        input_data,
        processing_instance_type,
        processing_instance_count,
        training_instance_type,
        training_instance_count,
        batch_instance_type,
        batch_instance_count,
    ],
    steps=[step_process, step_train, step_create_model],
    sagemaker_session=sess,
)
pipeline.upsert(role_arn=role)
execution = pipeline.start()
execution.wait()
step_create_model.properties.ModelName.__dict__<jupyter_output><empty_output><jupyter_text>Inference PipelineOnce our model has been trained and the model abstraction created, we can now define the pipeline that we will use to schedule our inferences. This is a very simple pipeline, only composed of one step, the `TransformStep`. In fact, this pipeline is used specifically to wrap around Batch Transform in order to run it on a schedule.<jupyter_code>from sagemaker.workflow.parameters import ParameterInteger, ParameterString

# model name
model_name = ParameterString(name="ModelName")

# test data
test_data = ParameterString(name="TestData")
output_path = ParameterString(name="OutputPath", default_value=f"s3://{bucket}/{prefix}/output/")

# batch inference step parameters
batch_instance_type = ParameterString(name="BatchInstanceType", default_value="ml.c5.xlarge")
batch_instance_count = ParameterInteger(name="BatchInstanceCount", default_value=1)
from sagemaker.sklearn import SKLearnModel
from sagemaker.transformer import Transformer

transformer = Transformer(
    model_name=model_name,
    instance_count=batch_instance_count,
    instance_type=batch_instance_type,
    base_transform_job_name="sklearn-transformer",
    output_path=output_path,
)
from sagemaker.workflow.steps import TransformStep
from sagemaker.inputs import TransformInput

transformer_step = TransformStep(
    name="scheduled-pipeline-sklearn-transformer",
    transformer=transformer,
    inputs=TransformInput(data=test_data, content_type="text/csv"),
)
from sagemaker.workflow.pipeline import Pipeline

pipeline_name = "sklearn-scheduled-inference"

pipeline = Pipeline(
    name=pipeline_name,
    parameters=[
        model_name,
        test_data,
        output_path,
        batch_instance_type,
        batch_instance_count,
    ],
    steps=[transformer_step],
    sagemaker_session=sess,
)
import json
from IPython.display import JSON

definition = json.loads(pipeline.definition())
JSON(definition)
pipeline.upsert(
    role_arn=role,
    description="A SM Pipeline to have scheduled inference with SM Batch Transform",
)
execution = pipeline.start(
    parameters={
        "ModelName": step_create_model.properties.ModelName,  # or replace with your model name
        "TestData": x_test_path,  # or replace with your S3 path
    }
)
execution.wait()
execution.describe()
!aws s3 ls s3://$bucket/$prefix/output/<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08531746031746032, 'fraction_numerical': 0.004238816738816739, 'mean_word_length': 5.443346891342243, 'pattern_counts': {'":': 9, '<': 33, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '427129', 'license': 'Apache-2.0', 'n_tokens_mistral': 3242, 'n_tokens_neox': 3131, 'n_words': 844, 'path': 'sagemaker-pipelines/tabular/sklearn-scheduled-training-and-inference/scheduled-sm-pipelines-sklearn.ipynb', 'repo_name': 'dgallitelli/amazon-sagemaker-examples', 'size': '11088'}