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 json\n",
    "from datetime import datetime\n",
    "\n",
    "from azureml.core import Workspace, ComputeTarget\n",
    "from azureml.core.authentication import InteractiveLoginAuthentication, ServicePrincipalAuthentication, \\\n",
    "    AzureCliAuthentication\n",
    "from azureml.core.compute import AmlCompute\n",
    "from azureml.core.datastore import Datastore\n",
    "from azureml.core.runconfig import CondaDependencies, RunConfiguration\n",
    "from azureml.data.data_reference import DataReference\n",
    "from azureml.exceptions import AuthenticationException\n",
    "from azureml.pipeline.core import Pipeline, PipelineData\n",
    "from azureml.pipeline.core.schedule import ScheduleRecurrence, Schedule\n",
    "from azureml.pipeline.steps import PythonScriptStep"
   ]
  },
  {
   "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": [
    "from dotenv import find_dotenv, get_key\n",
    "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."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_auth(env_path):\n",
    "    if get_key(env_path, 'password') != \"YOUR_SERVICE_PRINCIPAL_PASSWORD\":\n",
    "        aml_sp_password = get_key(env_path, 'password')\n",
    "        aml_sp_tennant_id = get_key(env_path, 'tenant_id')\n",
    "        aml_sp_username = get_key(env_path, 'username')\n",
    "        auth = ServicePrincipalAuthentication(\n",
    "            tenant_id=aml_sp_tennant_id,\n",
    "            service_principal_id=aml_sp_username,\n",
    "            service_principal_password=aml_sp_password\n",
    "        )\n",
    "    else:\n",
    "        try:\n",
    "            auth = AzureCliAuthentication()\n",
    "            auth.get_authentication_header()\n",
    "        except AuthenticationException:\n",
    "            auth = InteractiveLoginAuthentication()\n",
    "\n",
    "    return auth\n",
    "\n",
    "# Get AML workspace\n",
    "env_path = find_dotenv(raise_error_if_not_found=True)\n",
    "ws = Workspace.from_config(auth=get_auth(env_path))\n",
    "print(ws.name, ws.resource_group, ws.location, sep=\"\\n\")\n",
    "\n",
    "blob_datastore = ws.get_default_datastore()\n",
    "\n",
    "blob_account = blob_datastore.account_name\n",
    "blob_key = blob_datastore.account_key\n",
    "\n",
    "AML_VM_SIZE = \"Standard_D2\"\n",
    "AML_MIN_NODES = 2\n",
    "AML_MAX_NODES = 2"
   ]
  },
  {
   "cell_type": "markdown",
   "source": [
    "Create compute target"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "provisioning_config = AmlCompute.provisioning_configuration(vm_size = AML_VM_SIZE,\n",
    "                                                            min_nodes = AML_MIN_NODES,\n",
    "                                                            max_nodes = AML_MAX_NODES)\n",
    "\n",
    "compute_target = ComputeTarget.create(ws, j[\"cluster_name\"], provisioning_config)\n",
    "compute_target.wait_for_completion(show_output=True)"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "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",
    "    ws,\n",
    "    datastore_name=\"data_ds\",\n",
    "    container_name=j[\"data_blob_container\"],\n",
    "    account_name=blob_account,\n",
    "    account_key=blob_key,\n",
    ")\n",
    "data_dir = DataReference(datastore=data_ds, data_reference_name=\"data\")\n",
    "\n",
    "models_ds = Datastore.register_azure_blob_container(\n",
    "    ws,\n",
    "    datastore_name=\"models_ds\",\n",
    "    container_name=j[\"models_blob_container\"],\n",
    "    account_name=blob_account,\n",
    "    account_key=blob_key,\n",
    ")\n",
    "models_dir = DataReference(datastore=models_ds, data_reference_name=\"models\")\n",
    "\n",
    "preds_ds = Datastore.register_azure_blob_container(\n",
    "    ws,\n",
    "    datastore_name=\"preds_ds\",\n",
    "    container_name=j[\"preds_blob_container\"],\n",
    "    account_name=blob_account,\n",
    "    account_key=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=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=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"
  },
  "pycharm": {
   "stem_cell": {
    "cell_type": "raw",
    "source": [],
    "metadata": {
     "collapsed": false
    }
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}