   ml_ops/sm-pipelines_step_caching/sm-pipelines_step_caching.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "850a7792",
   "metadata": {},
   "source": [
    "# Use SageMaker Pipelines With Step Caching"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "af6144a0",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "This notebook's CI test result for us-west-2 is as follows. CI test results in other regions can be found at the end of the notebook. \n",
    "\n",
    "![This us-west-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/us-west-2/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "---"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "e0e3c54c",
   "metadata": {},
   "source": [
    "This notebook demonstrates how to take advantage of pipeline step caching. With step caching, SageMaker tracks the arguments used for each step execution and re-uses previous, successful executions when the call signatures match. SageMaker only tracks arguments important for the output of the step, so pipeline steps are optimized for cache hits and unnecessary step executions are avoided."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "3ed2bc66",
   "metadata": {},
   "source": [
    " See the [Developer Guide](https://docs.aws.amazon.com/sagemaker/latest/dg/pipelines-caching.html) and the [Python SDK docs](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#caching-configuration) for more information. "
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "6e9eb72e",
   "metadata": {},
   "source": [
    "## A SageMaker Pipeline\n",
    "The pipeline in this notebook follows a shortened version of a typical ML pattern. Just two steps are included - preprocessing and training."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "aa9349d0",
   "metadata": {},
   "source": [
    "## Dataset\n",
    "\n",
    "The dataset you use is the [UCI Machine Learning Abalone Dataset](https://archive.ics.uci.edu/ml/datasets/abalone) [1].  The aim for this task is to determine the age of an abalone snail from its physical measurements. At the core, this is a regression problem.\n",
    "\n",
    "The dataset contains several features: length (the longest shell measurement), diameter (the diameter perpendicular to length), height (the height with meat in the shell), whole_weight (the weight of whole abalone), shucked_weight (the weight of meat), viscera_weight (the gut weight after bleeding), shell_weight (the weight after being dried), sex ('M', 'F', 'I' where 'I' is Infant), and rings (integer).\n",
    "\n",
    "The number of rings turns out to be a good approximation for age (age is rings + 1.5). However, to obtain this number requires cutting the shell through the cone, staining the section, and counting the number of rings through a microscope, which is a time-consuming task. However, the other physical measurements are easier to determine. You use the dataset to build a predictive model of the variable rings through these other physical measurements.\n",
    "\n",
    "Before you upload the data to an S3 bucket, install the SageMaker Python SDK and gather some constants you can use later in this notebook.\n",
    "\n",
    "> [1] Dua, D. and Graff, C. (2019). [UCI Machine Learning Repository](http://archive.ics.uci.edu/ml). Irvine, CA: University of California, School of Information and Computer Science."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "14b47119",
   "metadata": {},
   "source": [
    "#### Install the latest version of the SageMaker Python SDK. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "06ebd8e5",
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install 'sagemaker' --upgrade"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "aed8ecd1",
   "metadata": {},
   "source": [
    "## Define Constants\n",
    "\n",
    "Before downloading the dataset, gather some constants you can use later in this notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3a70c4f6",
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "import sagemaker\n",
    "from sagemaker.workflow.pipeline_context import PipelineSession\n",
    "from sagemaker.workflow.steps import CacheConfig\n",
    "\n",
    "sagemaker_session = sagemaker.session.Session()\n",
    "region = sagemaker_session.boto_region_name\n",
    "role = sagemaker.get_execution_role()  # Or a literal role ARN you've created in your account\n",
    "pipeline_session = PipelineSession()\n",
    "default_bucket = sagemaker_session.default_bucket()\n",
    "default_bucket_prefix = sagemaker_session.default_bucket_prefix\n",
    "default_bucket_prefix_path = \"\"\n",
    "\n",
    "# If a default bucket prefix is specified, append it to the s3 path\n",
    "if default_bucket_prefix:\n",
    "    default_bucket_prefix_path = f\"/{default_bucket_prefix}\"\n",
    "\n",
    "model_package_group_name = f\"AbaloneModelPackageGroupName\"\n",
    "step_cache_config = CacheConfig(enable_caching=True, expire_after=\"T12H\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "37b897cf",
   "metadata": {},
   "source": [
    "Download the Abalone dataset:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "03487ce6",
   "metadata": {},
   "outputs": [],
   "source": [
    "local_path = \"artifacts/abalone-dataset.csv\"\n",
    "\n",
    "s3 = boto3.resource(\"s3\")\n",
    "s3.Bucket(f\"sagemaker-example-files-prod-{region}\").download_file(\n",
    "    \"datasets/tabular/uci_abalone/abalone.csv\", local_path\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "d7c848c8",
   "metadata": {},
   "source": [
    "## Define Parameters to Parametrize Pipeline Execution\n",
    "\n",
    "Define Pipeline parameters that you can use to parametrize the pipeline. Parameters enable custom pipeline executions and schedules without having to modify the Pipeline definition.\n",
    "\n",
    "The supported parameter types include:\n",
    "\n",
    "* `ParameterString` - represents a `str` Python type\n",
    "* `ParameterInteger` - represents an `int` Python type\n",
    "* `ParameterFloat` - represents a `float` Python type\n",
    "\n",
    "These parameters support providing a default value, which can be overridden on pipeline execution. The default value specified should be an instance of the type of the parameter.\n",
    "\n",
    "The parameters defined in this workflow include:\n",
    "\n",
    "* `processing_instance_count` - The instance count of the processing job.\n",
    "* `instance_type` - The `ml.*` instance type of the training job."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1c133653",
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.parameters import (\n",
    "    ParameterInteger,\n",
    "    ParameterString,\n",
    "    ParameterFloat,\n",
    ")\n",
    "\n",
    "processing_instance_count = ParameterInteger(name=\"ProcessingInstanceCount\", default_value=1)\n",
    "instance_type = ParameterString(name=\"TrainingInstanceType\", default_value=\"ml.m5.xlarge\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "f21ab6c4",
   "metadata": {},
   "source": [
    "## Define a Processing Step for Feature Engineering\n",
    "\n",
    "First, develop a preprocessing script that is specified in the Processing step.\n",
    "\n",
    "The file `preprocessing.py` in `artifacts/code` contains the preprocessing script. You can update the script and save the file to overwrite. The preprocessing script uses `scikit-learn` to do the following:\n",
    "\n",
    "* Fill in missing sex category data and encode it so that it is suitable for training.\n",
    "* Scale and normalize all numerical fields, aside from sex and rings numerical data.\n",
    "* Split the data into training, validation, and test datasets.\n",
    "\n",
    "The Processing step executes the script on the input data. The Training step uses the preprocessed training features and labels to train a model."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "a4160d00",
   "metadata": {},
   "source": [
    "Next, create an instance of a `SKLearnProcessor` processor and use that in our `ProcessingStep`.\n",
    "\n",
    "You also specify the `framework_version` to use throughout this notebook.\n",
    "\n",
    "Note the `processing_instance_count` parameter used by the processor instance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dcbf43c6",
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.sklearn.processing import SKLearnProcessor\n",
    "\n",
    "\n",
    "framework_version = \"0.23-1\"\n",
    "\n",
    "sklearn_processor = SKLearnProcessor(\n",
    "    framework_version=framework_version,\n",
    "    instance_type=\"ml.m5.xlarge\",\n",
    "    instance_count=processing_instance_count,\n",
    "    base_job_name=\"sklearn-abalone-process\",\n",
    "    role=role,\n",
    "    sagemaker_session=pipeline_session,\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "a45e80ec",
   "metadata": {},
   "source": [
    "Finally, we take the output of the processor's `run` method and pass that as arguments to the `ProcessingStep`. When passing a `pipeline_session` as the `sagemaker_session` parameter, this causes the `.run()` method to return a function call rather than launch a processing job. The function call executes once the pipeline gets built, and creates the arguments needed to run the job as a step in the pipeline.\n",
    "\n",
    "Note the `\"train\"` and `\"validation\"`, and `\"test\"` named channels specified in the output configuration for the processing job. Step `Properties` can be used in subsequent steps and resolve to their runtime values at execution. Specifically, this usage is called out when you define the training step."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fd4c40ad",
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.processing import ProcessingInput, ProcessingOutput\n",
    "from sagemaker.workflow.steps import ProcessingStep\n",
    "from sagemaker.dataset_definition.inputs import S3Input\n",
    "\n",
    "processor_args = sklearn_processor.run(\n",
    "    inputs=[\n",
    "        ProcessingInput(\n",
    "            source=\"artifacts/abalone-dataset.csv\",\n",
    "            input_name=\"abalone-dataset\",\n",
    "            s3_input=S3Input(\n",
    "                local_path=\"/opt/ml/processing/input\",\n",
    "                s3_uri=\"artifacts/abalone-dataset.csv\",\n",
    "                s3_data_type=\"S3Prefix\",\n",
    "                s3_input_mode=\"File\",\n",
    "                s3_data_distribution_type=\"FullyReplicated\",\n",
    "                s3_compression_type=\"None\",\n",
    "            ),\n",
    "        )\n",
    "    ],\n",
    "    outputs=[\n",
    "        ProcessingOutput(output_name=\"train\", source=\"/opt/ml/processing/train\"),\n",
    "        ProcessingOutput(output_name=\"validation\", source=\"/opt/ml/processing/validation\"),\n",
    "        ProcessingOutput(output_name=\"test\", source=\"/opt/ml/processing/test\"),\n",
    "    ],\n",
    "    code=\"artifacts/code/processing/preprocessing.py\",\n",
    ")\n",
    "\n",
    "step_process = ProcessingStep(\n",
    "    name=\"AbaloneProcess\", step_args=processor_args, cache_config=step_cache_config\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "6590afed",
   "metadata": {},
   "source": [
    "## Define a Training Step to Train a Model\n",
    "\n",
    "In this section, use Amazon SageMaker's [XGBoost Algorithm](https://docs.aws.amazon.com/sagemaker/latest/dg/xgboost.html) to train on this dataset. Configure an Estimator for the XGBoost algorithm and the input dataset. A typical training script loads data from the input channels, configures training with hyperparameters, trains a model, and saves a model to `model_dir` so that it can be hosted later.\n",
    "\n",
    "The model path where the models from training are saved is also specified.\n",
    "\n",
    "Note the `instance_type` parameter may be used in multiple places in the pipeline. In this case, the `instance_type` is passed into the estimator."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f9733437",
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.estimator import Estimator\n",
    "from sagemaker.inputs import TrainingInput\n",
    "from sagemaker.workflow.steps import TrainingStep\n",
    "\n",
    "model_path = f\"s3://{default_bucket}{default_bucket_prefix_path}/AbaloneTrain\"\n",
    "image_uri = sagemaker.image_uris.retrieve(\n",
    "    framework=\"xgboost\",\n",
    "    region=region,\n",
    "    version=\"1.0-1\",\n",
    "    py_version=\"py3\",\n",
    "    instance_type=\"ml.m5.xlarge\",\n",
    ")\n",
    "xgb_train = Estimator(\n",
    "    image_uri=image_uri,\n",
    "    instance_type=instance_type,\n",
    "    instance_count=1,\n",
    "    output_path=model_path,\n",
    "    role=role,\n",
    "    sagemaker_session=pipeline_session,\n",
    ")\n",
    "xgb_train.set_hyperparameters(\n",
    "    objective=\"reg:linear\",\n",
    "    num_round=50,\n",
    "    max_depth=5,\n",
    "    eta=0.2,\n",
    "    gamma=4,\n",
    "    min_child_weight=6,\n",
    "    subsample=0.7,\n",
    ")\n",
    "\n",
    "train_args = xgb_train.fit(\n",
    "    inputs={\n",
    "        \"train\": TrainingInput(\n",
    "            s3_data=step_process.properties.ProcessingOutputConfig.Outputs[\"train\"].S3Output.S3Uri,\n",
    "            content_type=\"text/csv\",\n",
    "        ),\n",
    "        \"validation\": TrainingInput(\n",
    "            s3_data=step_process.properties.ProcessingOutputConfig.Outputs[\n",
    "                \"validation\"\n",
    "            ].S3Output.S3Uri,\n",
    "            content_type=\"text/csv\",\n",
    "        ),\n",
    "    }\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "d74c6821",
   "metadata": {},
   "source": [
    "Finally, we use the output of the estimator's `.fit()` method as arguments to the `TrainingStep`. When passing a `pipeline_session` as the `sagemaker_session` parameter, this causes the `.fit()` method to return a function call rather than launch the training job. The function call executes once the pipeline gets built, and creates the arguments needed to run the job as a step in the pipeline.\n",
    "\n",
    "Pass in the `S3Uri` of the `\"train\"` output channel to the `.fit()` method. The `properties` attribute of a Pipeline step matches the object model of the corresponding response of a describe call. These properties can be referenced as placeholder values and are resolved at runtime. For example, the `ProcessingStep` `properties` attribute matches the object model of the [DescribeProcessingJob](https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_DescribeProcessingJob.html) response object."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "13a16305",
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.inputs import TrainingInput\n",
    "from sagemaker.workflow.steps import TrainingStep\n",
    "\n",
    "\n",
    "step_train = TrainingStep(name=\"AbaloneTrain\", step_args=train_args, cache_config=step_cache_config)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "4c3b2fd4",
   "metadata": {},
   "source": [
    "## Define a Pipeline of Parameters and Steps\n",
    "\n",
    "In this section, combine the steps into a Pipeline, so it can be executed.\n",
    "\n",
    "A pipeline requires a `name`, `parameters`, and `steps`. Names must be unique within an `(account, region)` pair.\n",
    "\n",
    "Note:\n",
    "\n",
    "* All the parameters used in the definitions must be present.\n",
    "* Steps passed into the pipeline do not have to be listed in the order of execution. The SageMaker Pipeline service resolves the data dependency DAG as steps for the execution to complete.\n",
    "* Steps must be unique to across the pipeline step list and all condition step if/else lists."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1e9f3194",
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.pipeline import Pipeline\n",
    "\n",
    "\n",
    "pipeline_name = f\"AbaloneBetaPipelineCaching\"\n",
    "pipeline = Pipeline(\n",
    "    name=pipeline_name,\n",
    "    parameters=[\n",
    "        processing_instance_count,\n",
    "        instance_type,\n",
    "    ],\n",
    "    steps=[step_process, step_train],\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "d894c320",
   "metadata": {},
   "source": [
    "### (Optional) Examining the pipeline definition\n",
    "\n",
    "The JSON of the pipeline definition can be examined to confirm the pipeline is well-defined and the parameters and step properties resolve correctly."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "673480a3",
   "metadata": {},
   "source": [
    "For example, you might check the `ProcessingInputs` of the pre-processing step. The Python SDK intentionally structures input code artifacts' S3 paths in order to optimize caching - more explanation on this later in the notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b74f9a26",
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "\n",
    "\n",
    "definition = json.loads(pipeline.definition())\n",
    "definition"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "97a47c14",
   "metadata": {},
   "source": [
    "## Submit the pipeline to SageMaker and start execution\n",
    "\n",
    "Submit the pipeline definition to the Pipeline service. The Pipeline service uses the role that is passed in to create all the jobs defined in the steps."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2a7c3955",
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline.upsert(role_arn=role)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "ad7a8e08",
   "metadata": {},
   "source": [
    "Start the pipeline and accept all the default parameters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2937c7ec",
   "metadata": {},
   "outputs": [],
   "source": [
    "execution = pipeline.start()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "6893d12b",
   "metadata": {},
   "source": [
    "## Pipeline Operations: Examining and Waiting for Pipeline Execution\n",
    "\n",
    "Describe the pipeline execution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f2c52869",
   "metadata": {},
   "outputs": [],
   "source": [
    "execution.describe()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "f08c06a5",
   "metadata": {},
   "source": [
    "Wait for the execution to complete."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "245502ee",
   "metadata": {},
   "outputs": [],
   "source": [
    "execution.wait()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "9643a048",
   "metadata": {},
   "source": [
    "List the steps in the execution. These are the steps in the pipeline that have been resolved by the step executor service."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ffe5c4ab",
   "metadata": {},
   "outputs": [],
   "source": [
    "execution.list_steps()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "dfbba5e2",
   "metadata": {},
   "source": [
    "## Caching Behavior\n",
    "In the next part of the notebook, we observe both cache hit and cache miss scenarios. There are many parameters that are passed into SageMaker pipeline steps. Some directly influence the results of the corresponding SageMaker jobs such as the input data, while others describe how the job will run, for example an `instance_type`. When parameters from the first group are updated, a cache miss occurs and the step re-runs. When parameters from the second group are updated, a cache hit occurs and the step does not execute, as the job results are unaffected. In the following pipeline execution examples, parameters from both categories are updated and the effects of each one are observed."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "8b287cd8",
   "metadata": {},
   "source": [
    "There are many other parameters outside of these examples - for more information on how they affect caching, or for more information on how to opt in to or out of caching, please refer to the [Developer Guide](https://docs.aws.amazon.com/sagemaker/latest/dg/pipelines-caching.html) and the [Python SDK docs](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#caching-configuration). "
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "89e04224",
   "metadata": {},
   "source": [
    "**Hint:** If you are executing this notebook in SageMaker Studio, use the following tip to easily track caching behavior."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "24d93baf",
   "metadata": {},
   "source": [
    "To verify whether a cache hit or miss occurred for a particular step during a pipeline execution, open the SageMaker resources tab on the left. Click on Pipelines in the dropdown menu and find the \"AbaloneBetaPipelineCaching\" pipeline created in this notebook. Click on the pipeline in order to view the different executions tracked under that pipeline. You can click on each execution to view a graph of the steps and their behavior during that execution. In the graph, click on a step and then click on the \"information\" column to view the cache information."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "e6b60061",
   "metadata": {},
   "source": [
    "Here is an example of a cache hit in SageMaker Studio, displayed in the pane on the right side of the page:"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "0df8e0cf",
   "metadata": {},
   "source": [
    "![\"studio cache hit image\"](artifacts/studio_cache_hit.png)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "adc61a47",
   "metadata": {},
   "source": [
    "And here is an example of a cache miss in SageMaker Studio:"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "e79ea02d",
   "metadata": {},
   "source": [
    "![\"studio cache miss image\"](artifacts/studio_cache_miss.png)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "ecaf5b1f",
   "metadata": {},
   "source": [
    "Information tab with cache hit result, enlarged:"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "fd313931",
   "metadata": {},
   "source": [
    "![\"studio cache hit zoomed image\"](artifacts/studio_cache_hit_zoomed.png)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "733c954f",
   "metadata": {},
   "source": [
    "### Cache Hit\n",
    "Now that the pipeline has executed, the cache for the steps has been created. To observe cache hit behavior, change the `instance_type` parameter for both steps, from xlarge to large."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e776bdb7",
   "metadata": {},
   "outputs": [],
   "source": [
    "sklearn_processor.instance_type = \"ml.m5.large\"\n",
    "xgb_train.instance_type = \"ml.m5.large\""
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "be73a872",
   "metadata": {},
   "source": [
    "Create the step args again, and pass the updated steps to the pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "275f5cf6",
   "metadata": {},
   "outputs": [],
   "source": [
    "processor_args = sklearn_processor.run(\n",
    "    inputs=[\n",
    "        ProcessingInput(\n",
    "            source=\"artifacts/abalone-dataset.csv\",\n",
    "            input_name=\"abalone-dataset\",\n",
    "            s3_input=S3Input(\n",
    "                local_path=\"/opt/ml/processing/input\",\n",
    "                s3_uri=\"artifacts/abalone-dataset.csv\",\n",
    "                s3_data_type=\"S3Prefix\",\n",
    "                s3_input_mode=\"File\",\n",
    "                s3_data_distribution_type=\"FullyReplicated\",\n",
    "                s3_compression_type=\"None\",\n",
    "            ),\n",
    "        )\n",
    "    ],\n",
    "    outputs=[\n",
    "        ProcessingOutput(output_name=\"train\", source=\"/opt/ml/processing/train\"),\n",
    "        ProcessingOutput(output_name=\"validation\", source=\"/opt/ml/processing/validation\"),\n",
    "        ProcessingOutput(output_name=\"test\", source=\"/opt/ml/processing/test\"),\n",
    "    ],\n",
    "    code=\"artifacts/code/processing/preprocessing.py\",\n",
    ")\n",
    "\n",
    "step_process = ProcessingStep(\n",
    "    name=\"AbaloneProcess\", step_args=processor_args, cache_config=step_cache_config\n",
    ")\n",
    "\n",
    "train_args = xgb_train.fit(\n",
    "    inputs={\n",
    "        \"train\": TrainingInput(\n",
    "            s3_data=step_process.properties.ProcessingOutputConfig.Outputs[\"train\"].S3Output.S3Uri,\n",
    "            content_type=\"text/csv\",\n",
    "        ),\n",
    "        \"validation\": TrainingInput(\n",
    "            s3_data=step_process.properties.ProcessingOutputConfig.Outputs[\n",
    "                \"validation\"\n",
    "            ].S3Output.S3Uri,\n",
    "            content_type=\"text/csv\",\n",
    "        ),\n",
    "    }\n",
    ")\n",
    "\n",
    "step_train = TrainingStep(name=\"AbaloneTrain\", step_args=train_args, cache_config=step_cache_config)\n",
    "\n",
    "pipeline.steps = [step_process, step_train]"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "5f2eea68",
   "metadata": {},
   "source": [
    "View the pipeline definition again and verify our changes are reflected there."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dea4a7c6",
   "metadata": {},
   "outputs": [],
   "source": [
    "definition = json.loads(pipeline.definition())\n",
    "definition"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "8a8d759d",
   "metadata": {},
   "source": [
    "Update the pipeline and re-execute. The new execution results in cache hits for both steps, as the `instance_type` parameter does not affect the result of the jobs. SageMaker does not track this parameter when evaluating the cache for previous step executions, so it has no effect."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "769b65b0",
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline.update(role)\n",
    "second_execution = pipeline.start()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "dd02a8d0",
   "metadata": {},
   "source": [
    "Describe the new execution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bfd5995a",
   "metadata": {},
   "outputs": [],
   "source": [
    "second_execution.describe()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "5296477c",
   "metadata": {},
   "source": [
    "Wait for the new execution to complete."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1414d9dd",
   "metadata": {},
   "outputs": [],
   "source": [
    "second_execution.wait()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "aff697d1",
   "metadata": {},
   "source": [
    "List the steps in the new execution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "522dd73a",
   "metadata": {},
   "outputs": [],
   "source": [
    "second_execution.list_steps()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "bcce7276",
   "metadata": {},
   "source": [
    "### Cache Miss"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "6b26de1c",
   "metadata": {},
   "source": [
    "Now, change a different set of parameters for the steps. For the processing step, use a different code script from the artifacts directory. For the training step, update some hyperparameters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9fc5b7d9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# processing\n",
    "processor_args = sklearn_processor.run(\n",
    "    inputs=[\n",
    "        ProcessingInput(\n",
    "            source=\"artifacts/abalone-dataset.csv\",\n",
    "            input_name=\"abalone-dataset\",\n",
    "            s3_input=S3Input(\n",
    "                local_path=\"/opt/ml/processing/input\",\n",
    "                s3_uri=\"artifacts/abalone-dataset.csv\",\n",
    "                s3_data_type=\"S3Prefix\",\n",
    "                s3_input_mode=\"File\",\n",
    "                s3_data_distribution_type=\"FullyReplicated\",\n",
    "                s3_compression_type=\"None\",\n",
    "            ),\n",
    "        )\n",
    "    ],\n",
    "    outputs=[\n",
    "        ProcessingOutput(output_name=\"train\", source=\"/opt/ml/processing/train\"),\n",
    "        ProcessingOutput(output_name=\"validation\", source=\"/opt/ml/processing/validation\"),\n",
    "        ProcessingOutput(output_name=\"test\", source=\"/opt/ml/processing/test\"),\n",
    "    ],\n",
    "    code=\"artifacts/code/processing/preprocessing_2.py\",\n",
    ")\n",
    "\n",
    "step_process = ProcessingStep(\n",
    "    name=\"AbaloneProcess\", step_args=processor_args, cache_config=step_cache_config\n",
    ")\n",
    "\n",
    "\n",
    "# training\n",
    "xgb_train.set_hyperparameters(\n",
    "    objective=\"reg:linear\",\n",
    "    num_round=30,\n",
    "    max_depth=4,\n",
    "    eta=0.2,\n",
    "    gamma=5,\n",
    "    min_child_weight=6,\n",
    "    subsample=0.6,\n",
    ")\n",
    "\n",
    "train_args = xgb_train.fit(\n",
    "    inputs={\n",
    "        \"train\": TrainingInput(\n",
    "            s3_data=step_process.properties.ProcessingOutputConfig.Outputs[\"train\"].S3Output.S3Uri,\n",
    "            content_type=\"text/csv\",\n",
    "        ),\n",
    "        \"validation\": TrainingInput(\n",
    "            s3_data=step_process.properties.ProcessingOutputConfig.Outputs[\n",
    "                \"validation\"\n",
    "            ].S3Output.S3Uri,\n",
    "            content_type=\"text/csv\",\n",
    "        ),\n",
    "    }\n",
    ")\n",
    "\n",
    "step_train = TrainingStep(name=\"AbaloneTrain\", step_args=train_args, cache_config=step_cache_config)\n",
    "\n",
    "pipeline.steps = [step_process, step_train]"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "c829e08c",
   "metadata": {},
   "source": [
    "View the pipeline definition again and verify the changes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2f589ec1",
   "metadata": {},
   "outputs": [],
   "source": [
    "definition = json.loads(pipeline.definition())\n",
    "definition"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "06bbd71d",
   "metadata": {},
   "source": [
    "Because input code artifacts and hyperparameters directly affect the job results, these attributes are tracked by SageMaker. This results in cache misses during the next pipeline execution, and both steps re-execute."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "ef19b698",
   "metadata": {},
   "source": [
    "**Note**: When local data or code artifacts are passed in as parameters to pipeline steps, the Python SDK uses a specific path structure when uploading these artifacts to S3. The contents of code files and in some cases configuration files are hashed, and this hash is included in the S3 upload path (View the pipeline definition to see the path structure). Because SageMaker tracks the S3 paths of these artifacts when evaluating whether a step has already executed or not, this ensures that when a new local code or data file is provided, the SDK creates a new S3 upload path, a cache miss will occur, and the step will run again with the new data. For more information on the Python SDK's S3 path structures, see the [Python SDK docs](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#caching-configuration)."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "cab07c80",
   "metadata": {},
   "source": [
    "Update the pipeline and re-execute."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4f089694",
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline.update(role)\n",
    "third_execution = pipeline.start()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "4da94c1e",
   "metadata": {},
   "source": [
    "Describe the new execution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "97882392",
   "metadata": {},
   "outputs": [],
   "source": [
    "third_execution.describe()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "667556c7",
   "metadata": {},
   "source": [
    "Wait for the new execution to complete."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8eb5f847",
   "metadata": {},
   "outputs": [],
   "source": [
    "third_execution.wait()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "c5be8ed6",
   "metadata": {},
   "source": [
    "List the steps in the new execution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8d73b2e9",
   "metadata": {},
   "outputs": [],
   "source": [
    "third_execution.list_steps()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "c83253a3",
   "metadata": {},
   "source": [
    "## Notebook CI Test Results\n",
    "\n",
    "This notebook was tested in multiple regions. The test results are as follows, except for us-west-2 which is shown at the top of the notebook.\n",
    "\n",
    "![This us-east-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/us-east-1/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This us-east-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/us-east-2/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This us-west-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/us-west-1/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This ca-central-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ca-central-1/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This sa-east-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/sa-east-1/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This eu-west-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-west-1/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This eu-west-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-west-2/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This eu-west-3 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-west-3/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This eu-central-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-central-1/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This eu-north-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-north-1/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This ap-southeast-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-southeast-1/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This ap-southeast-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-southeast-2/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This ap-northeast-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-northeast-1/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This ap-northeast-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-northeast-2/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n",
    "\n",
    "![This ap-south-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-south-1/ml_ops|sm-pipelines_step_caching|sm-pipelines_step_caching.ipynb)\n"
   ]
  }
 ],
 "metadata": {
  "instance_type": "ml.t3.medium",
  "kernelspec": {
   "display_name": "Python 3 (Data Science 3.0)",
   "language": "python",
   "name": "python3__SAGEMAKER_INTERNAL__arn:aws:sagemaker:us-east-1:081325390199:image/sagemaker-data-science-310-v1"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}