02_create_pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating and Scheduling the AML Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import json\n",
    "import sys\n",
    "from datetime import datetime\n",
    "from azureml.core.compute import AmlCompute\n",
    "from azureml.core.datastore import Datastore\n",
    "from azureml.data.data_reference import DataReference\n",
    "from azureml.pipeline.core import Pipeline, PipelineData\n",
    "from azureml.pipeline.steps import PythonScriptStep\n",
    "from azureml.core.runconfig import CondaDependencies, RunConfiguration\n",
    "from azureml.core import Workspace, Run, Experiment\n",
    "from azureml.core.authentication import InteractiveLoginAuthentication\n",
    "from azureml.pipeline.core.schedule import ScheduleRecurrence, Schedule"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Read the pipeline configuration parameters that were created in the previous notebook:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_config = \"pipeline_config.json\"\n",
    "with open(pipeline_config) as f:\n",
    "    j = json.loads(f.read())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Retrieve the AML workspace and compute target that were created previously:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Authenticate\n",
    "auth = InteractiveLoginAuthentication()\n",
    "\n",
    "# AML workspace\n",
    "aml_ws = Workspace.get(\n",
    "    name=j[\"aml_work_space\"],\n",
    "    auth=auth,\n",
    "    subscription_id=str(j[\"subscription_id\"]),\n",
    "    resource_group=j[\"resource_group_name\"],\n",
    ")\n",
    "\n",
    "# AML compute target\n",
    "compute_target = AmlCompute(aml_ws, j[\"cluster_name\"])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create the scoring pipeline's input and output by registering the Azure blob containers that were created:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Pipeline input and output\n",
    "data_ds = Datastore.register_azure_blob_container(\n",
    "    aml_ws,\n",
    "    datastore_name=\"data_ds\",\n",
    "    container_name=j[\"data_blob_container\"],\n",
    "    account_name=j[\"blob_account\"],\n",
    "    account_key=j[\"blob_key\"],\n",
    ")\n",
    "data_dir = DataReference(datastore=data_ds, data_reference_name=\"data\")\n",
    "\n",
    "models_ds = Datastore.register_azure_blob_container(\n",
    "    aml_ws,\n",
    "    datastore_name=\"models_ds\",\n",
    "    container_name=j[\"models_blob_container\"],\n",
    "    account_name=j[\"blob_account\"],\n",
    "    account_key=j[\"blob_key\"],\n",
    ")\n",
    "models_dir = DataReference(datastore=models_ds, data_reference_name=\"models\")\n",
    "\n",
    "preds_ds = Datastore.register_azure_blob_container(\n",
    "    aml_ws,\n",
    "    datastore_name=\"preds_ds\",\n",
    "    container_name=j[\"preds_blob_container\"],\n",
    "    account_name=j[\"blob_account\"],\n",
    "    account_key=j[\"blob_key\"],\n",
    ")\n",
    "preds_dir = PipelineData(name=\"preds\", datastore=preds_ds, is_directory=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Define the run configuration of the scoring pipeline by specifying the Python dependencies and enabling a Docker containerized execution environment:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run config\n",
    "conda_dependencies = CondaDependencies.create(\n",
    "    pip_packages=j[\"pip_packages\"], python_version=j[\"python_version\"]\n",
    ")\n",
    "run_config = RunConfiguration(conda_dependencies=conda_dependencies)\n",
    "run_config.environment.docker.enabled = True"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a scoring pipeline with multiple steps. Each step will execute the scoring Python script with different arguments, corresponding to a specific sensor. The steps will run independently in parallel and their execution will be managed by AML."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a pipeline step for each (device, sensor) pair\n",
    "steps = []\n",
    "for device_id in j[\"device_ids\"]:\n",
    "    for sensor in j[\"sensors\"]:\n",
    "        preds_dir = PipelineData(name=\"preds\", datastore=preds_ds, is_directory=True)\n",
    "        step = PythonScriptStep(\n",
    "            name=\"{}_{}\".format(device_id, sensor),\n",
    "            script_name=j[\"python_script_name\"],\n",
    "            arguments=[device_id, sensor, models_dir, data_dir, j[\"data_blob\"], preds_dir],\n",
    "            inputs=[models_dir, data_dir],\n",
    "            outputs=[preds_dir],\n",
    "            source_directory=j[\"python_script_directory\"],\n",
    "            compute_target=compute_target,\n",
    "            runconfig=run_config,\n",
    "            allow_reuse=False,\n",
    "        )\n",
    "        steps.append(step)\n",
    "\n",
    "pipeline = Pipeline(workspace=aml_ws, steps=steps)\n",
    "pipeline.validate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Publish the pipeline so that it becomes available for scheduling. Publishing a pipeline also allows deploying it as a web service."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Publish pipeline\n",
    "pipeline_name = \"scoring_pipeline_{}\".format(datetime.now().strftime(\"%y%m%d%H%M\"))\n",
    "published_pipeline = pipeline.publish(name=pipeline_name, description=pipeline_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Schedule the pipeline to run on the specified frequency and interval. This will also submit the initial pipeline run, as we don't specify any starting time for the schedule."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Schedule pipeline\n",
    "experiment_name = \"exp_\" + datetime.now().strftime(\"%y%m%d%H%M%S\")\n",
    "recurrence = ScheduleRecurrence(frequency=j['sched_frequency'], interval=j['sched_interval'])\n",
    "schedule = Schedule.create(\n",
    "    workspace=aml_ws,\n",
    "    name=\"{}_sched\".format(j[\"resource_group_name\"]),\n",
    "    pipeline_id=published_pipeline.id,\n",
    "    experiment_name=experiment_name,\n",
    "    recurrence=recurrence,\n",
    "    description=\"{}_sched\".format(j[\"resource_group_name\"]),\n",
    ")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python [conda env:amlmm]",
   "language": "python",
   "name": "conda-env-amlmm-py"
  },
  "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}