v1/python-sdk/tutorials/using-pipelines/publish-and-run-using-rest-endpoint.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# How to Publish a Pipeline and Invoke the REST endpoint\n",
    "In this notebook, we will see how we can publish a pipeline and then invoke the REST endpoint.\n",
    "\n",
    "### Initialization Steps"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618429162484
    }
   },
   "outputs": [],
   "source": [
    "import azureml.core\n",
    "from azureml.core import Workspace, Datastore, Experiment, Dataset\n",
    "from azureml.data import OutputFileDatasetConfig\n",
    "from azureml.core.compute import AmlCompute\n",
    "from azureml.core.compute import ComputeTarget\n",
    "\n",
    "# Check core SDK version number\n",
    "print(\"SDK version:\", azureml.core.VERSION)\n",
    "\n",
    "from azureml.pipeline.core import Pipeline\n",
    "from azureml.pipeline.steps import PythonScriptStep\n",
    "from azureml.pipeline.core.graph import PipelineParameter\n",
    "\n",
    "print(\"Pipeline SDK-specific imports completed\")\n",
    "\n",
    "ws = Workspace.from_config()\n",
    "print(ws.name, ws.resource_group, ws.location, ws.subscription_id, sep=\"\\n\")\n",
    "\n",
    "# Default datastore (Azure blob storage)\n",
    "# def_blob_store = ws.get_default_datastore()\n",
    "def_blob_store = Datastore(ws, \"workspaceblobstore\")\n",
    "print(\"Blobstore's name: {}\".format(def_blob_store.name))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Compute Targets\n",
    "#### Retrieve an already attached  Azure Machine Learning Compute"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618436520233
    }
   },
   "outputs": [],
   "source": [
    "from azureml.core.compute_target import ComputeTargetException\n",
    "\n",
    "# Choose a name for your cluster.\n",
    "amlcompute_cluster_name = \"cpu-cluster\"\n",
    "\n",
    "found = False\n",
    "# Check if this compute target already exists in the workspace.\n",
    "cts = ws.compute_targets\n",
    "if amlcompute_cluster_name in cts and cts[amlcompute_cluster_name].type == \"AmlCompute\":\n",
    "    found = True\n",
    "    print(\"Found existing compute target.\")\n",
    "    aml_compute = cts[amlcompute_cluster_name]\n",
    "\n",
    "if not found:\n",
    "    print(\"Creating a new compute target...\")\n",
    "    provisioning_config = AmlCompute.provisioning_configuration(\n",
    "        vm_size=\"STANDARD_D2_V2\",  # for GPU, use \"STANDARD_NC6s_v3\"\n",
    "        # vm_priority = 'lowpriority', # optional\n",
    "        max_nodes=4,\n",
    "    )\n",
    "\n",
    "    # Create the cluster.\n",
    "    aml_compute = ComputeTarget.create(ws, amlcompute_cluster_name, provisioning_config)\n",
    "\n",
    "    # Can poll for a minimum number of nodes and for a specific timeout.\n",
    "    # If no min_node_count is provided, it will use the scale settings for the cluster.\n",
    "    aml_compute.wait_for_completion(show_output=True, timeout_in_minutes=10)\n",
    "\n",
    "    # For a more detailed view of current AmlCompute status, use get_status()."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618433416002
    }
   },
   "outputs": [],
   "source": [
    "# For a more detailed view of current Azure Machine Learning Compute status, use get_status()\n",
    "# example: un-comment the following line.\n",
    "# print(aml_compute.get_status().serialize())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Building Pipeline Steps with Inputs and Outputs\n",
    "A step in the pipeline can take dataset as input. This dataset can be a data source that lives in one of the accessible data locations, or intermediate data produced by a previous step in the pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618433422307
    }
   },
   "outputs": [],
   "source": [
    "# Specify a public dataset path\n",
    "data_path = \"https://dprepdata.blob.core.windows.net/demo/Titanic.csv\"\n",
    "# Or uploading data to the datastore\n",
    "# data_path = def_blob_store.upload_files([\"./your_data.pkl\"], target_path=\"your_path\", overwrite=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618436536079
    }
   },
   "outputs": [],
   "source": [
    "# Reference the data uploaded to blob storage using file dataset\n",
    "# Assign the datasource to blob_input_data variable\n",
    "blob_input_data = Dataset.File.from_files(data_path).as_named_input(\"test_data\")\n",
    "print(\"Dataset created\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618436540369
    }
   },
   "outputs": [],
   "source": [
    "# Define intermediate data using OutputFileDatasetConfig\n",
    "processed_data1 = OutputFileDatasetConfig(name=\"processed_data1\")\n",
    "print(\"Output dataset object created\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Define a Step that consumes a dataset and produces intermediate data.\n",
    "In this step, we define a step that consumes a dataset and produces intermediate data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618436550106
    }
   },
   "outputs": [],
   "source": [
    "# trainStep consumes the datasource (Datareference) in the previous step\n",
    "# and produces processed_data1\n",
    "\n",
    "source_directory = \"publish_run_train\"\n",
    "\n",
    "trainStep = PythonScriptStep(\n",
    "    script_name=\"train.py\",\n",
    "    arguments=[\n",
    "        \"--input_data\",\n",
    "        blob_input_data.as_mount(),\n",
    "        \"--output_train\",\n",
    "        processed_data1,\n",
    "    ],\n",
    "    compute_target=aml_compute,\n",
    "    source_directory=source_directory,\n",
    ")\n",
    "print(\"trainStep created\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Define a Step that consumes intermediate data and produces intermediate data\n",
    "In this step, we define a step that consumes an intermediate data and produces intermediate data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618436553493
    }
   },
   "outputs": [],
   "source": [
    "# extractStep to use the intermediate data produced by trainStep\n",
    "# This step also produces an output processed_data2\n",
    "processed_data2 = OutputFileDatasetConfig(name=\"processed_data2\")\n",
    "source_directory = \"publish_run_extract\"\n",
    "\n",
    "extractStep = PythonScriptStep(\n",
    "    script_name=\"extract.py\",\n",
    "    arguments=[\n",
    "        \"--input_extract\",\n",
    "        processed_data1.as_input(),\n",
    "        \"--output_extract\",\n",
    "        processed_data2,\n",
    "    ],\n",
    "    compute_target=aml_compute,\n",
    "    source_directory=source_directory,\n",
    ")\n",
    "print(\"extractStep created\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Define a Step that consumes multiple intermediate data and produces intermediate data\n",
    "In this step, we define a step that consumes multiple intermediate data and produces intermediate data."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### PipelineParameter\n",
    "This step also has a PipelineParameter argument that help with calling the REST endpoint of the published pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618436557940
    }
   },
   "outputs": [],
   "source": [
    "# We will use this later in publishing pipeline\n",
    "pipeline_param = PipelineParameter(name=\"pipeline_arg\", default_value=10)\n",
    "print(\"pipeline parameter created\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Open `compare.py` in the local machine and examine the arguments, inputs, and outputs for the script. That will give you a good sense of why the script argument names used below are important.**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618436567924
    }
   },
   "outputs": [],
   "source": [
    "# Now define compareStep that takes two inputs (both intermediate data), and produce an output\n",
    "processed_data3 = OutputFileDatasetConfig(name=\"processed_data3\")\n",
    "\n",
    "# You can register the output as dataset after job completion\n",
    "processed_data3 = processed_data3.register_on_complete(\"compare_result\")\n",
    "\n",
    "source_directory = \"publish_run_compare\"\n",
    "\n",
    "compareStep = PythonScriptStep(\n",
    "    script_name=\"compare.py\",\n",
    "    arguments=[\n",
    "        \"--compare_data1\",\n",
    "        processed_data1.as_input(),\n",
    "        \"--compare_data2\",\n",
    "        processed_data2.as_input(),\n",
    "        \"--output_compare\",\n",
    "        processed_data3,\n",
    "        \"--pipeline_param\",\n",
    "        pipeline_param,\n",
    "    ],\n",
    "    compute_target=aml_compute,\n",
    "    source_directory=source_directory,\n",
    ")\n",
    "print(\"compareStep created\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Build the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618436696567
    }
   },
   "outputs": [],
   "source": [
    "pipeline1 = Pipeline(workspace=ws, steps=[compareStep])\n",
    "print(\"Pipeline is built\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Run published pipeline\n",
    "### Publish the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618436782680
    }
   },
   "outputs": [],
   "source": [
    "published_pipeline1 = pipeline1.publish(\n",
    "    name=\"My_New_Pipeline\",\n",
    "    description=\"My Published Pipeline Description\",\n",
    "    continue_on_step_failure=True,\n",
    ")\n",
    "published_pipeline1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Publish the pipeline from a submitted PipelineRun\n",
    "It is also possible to publish a pipeline from a submitted PipelineRun"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618436790538
    }
   },
   "outputs": [],
   "source": [
    "# submit a pipeline run\n",
    "pipeline_run1 = Experiment(ws, \"Pipeline_experiment\").submit(pipeline1)\n",
    "# publish a pipeline from the submitted pipeline run\n",
    "published_pipeline2 = pipeline_run1.publish_pipeline(\n",
    "    name=\"My_New_Pipeline2\",\n",
    "    description=\"My Published Pipeline Description\",\n",
    "    version=\"0.1\",\n",
    "    continue_on_step_failure=True,\n",
    ")\n",
    "published_pipeline2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Get published pipeline\n",
    "\n",
    "You can get the published pipeline using **pipeline id**.\n",
    "\n",
    "To get all the published pipelines for a given workspace(ws): \n",
    "```css\n",
    "all_pub_pipelines = PublishedPipeline.get_all(ws)\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618436798773
    }
   },
   "outputs": [],
   "source": [
    "from azureml.pipeline.core import PublishedPipeline\n",
    "\n",
    "pipeline_id = published_pipeline1.id  # use your published pipeline id\n",
    "published_pipeline = PublishedPipeline.get(ws, pipeline_id)\n",
    "published_pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Run published pipeline using its REST endpoint\n",
    "[This notebook](https://aka.ms/pl-restep-auth) shows how to authenticate to AML workspace."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618436804289
    }
   },
   "outputs": [],
   "source": [
    "from azureml.core.authentication import InteractiveLoginAuthentication\n",
    "import requests\n",
    "\n",
    "auth = InteractiveLoginAuthentication()\n",
    "aad_token = auth.get_authentication_header()\n",
    "\n",
    "rest_endpoint = published_pipeline.endpoint\n",
    "\n",
    "print(\n",
    "    \"You can perform HTTP POST on URL {} to trigger this pipeline\".format(rest_endpoint)\n",
    ")\n",
    "\n",
    "# specify the param when running the pipeline\n",
    "response = requests.post(\n",
    "    rest_endpoint,\n",
    "    headers=aad_token,\n",
    "    json={\n",
    "        \"ExperimentName\": \"My_Pipeline1\",\n",
    "        \"RunSource\": \"SDK\",\n",
    "        \"ParameterAssignments\": {\"pipeline_arg\": 45},\n",
    "    },\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1618436810043
    }
   },
   "outputs": [],
   "source": [
    "try:\n",
    "    response.raise_for_status()\n",
    "except Exception:\n",
    "    raise Exception(\n",
    "        \"Received bad response from the endpoint: {}\\n\"\n",
    "        \"Response Code: {}\\n\"\n",
    "        \"Headers: {}\\n\"\n",
    "        \"Content: {}\".format(\n",
    "            rest_endpoint, response.status_code, response.headers, response.content\n",
    "        )\n",
    "    )\n",
    "\n",
    "run_id = response.json().get(\"Id\")\n",
    "print(\"Submitted pipeline run: \", run_id)"
   ]
  }
 ],
 "metadata": {
  "authors": [
   {
    "name": "sanpil"
   }
  ],
  "category": "tutorial",
  "compute": [
   "AML Compute"
  ],
  "datasets": [
   "Custom"
  ],
  "deployment": [
   "None"
  ],
  "exclude_from_index": false,
  "framework": [
   "Azure ML"
  ],
  "friendly_name": "How to Publish a Pipeline and Invoke the REST endpoint",
  "kernel_info": {
   "name": "python38-azureml"
  },
  "kernelspec": {
   "display_name": "Python 3.8 - AzureML",
   "language": "python",
   "name": "python38-azureml"
  },
  "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.6.9"
  },
  "microsoft": {
   "host": {
    "AzureML": {
     "notebookHasBeenCompleted": true
    }
   }
  },
  "nteract": {
   "version": "nteract-front-end@1.0.0"
  },
  "order_index": 3,
  "star_tag": [
   "featured"
  ],
  "tags": [
   "None"
  ],
  "task": "Demonstrates the use of Published Pipelines"
 },
 "nbformat": 4,
 "nbformat_minor": 2
}