how-to-model-scope-exepool-experiment-using-aml-pipelines.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Copyright (c) Microsoft Corporation. All rights reserved.  \n",
    "Licensed under the MIT License."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Scope/ExePool Experiment using AML Pipelines\n",
    "This notebook will show how you can run a Scope job in a migrated ADLA account, and then run an Windows exe on the result of the Scope job using AML Pipeline."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Initialize Workspace\n",
    "\n",
    "Initialize a workspace object from persisted configuration. Make sure the config file is present at .\\config.json"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import azureml.core\n",
    "from azureml.core import Workspace, Run, Experiment\n",
    "from azureml.core.compute import ComputeTarget, DataFactoryCompute\n",
    "from azureml.core.datastore import Datastore\n",
    "from azureml.data.data_reference import DataReference\n",
    "from azureml.exceptions import ComputeTargetException\n",
    "from azureml.pipeline.core import Pipeline, PipelineData\n",
    "from azureml.pipeline.steps import AdlaStep, AzureBatchStep, DataTransferStep\n",
    "from azureml.pipeline.steps.azurebatch_step import AzureBatchTaskInfo\n",
    "from azureml.pipeline.steps_internal import ScopeStep\n",
    "\n",
    "print(\"SDK version:\", azureml.core.VERSION)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Account Details\n",
    "Get the details of the account to be used from this [OneNote page](https://microsoft.sharepoint.com/teams/azuremlnursery/_layouts/OneNote.aspx?id=%2Fteams%2Fazuremlnursery%2FSiteAssets%2FAzure%20ML%20Nursery%20Notebook&wd=target%28Workshop.one%7C265D85D5-44C8-9D40-B556-A31FA098E708%2FPipeline%3A%20Scope%20and%20Batch%7CE40F0043-079C-4551-BAD6-27E5D056D05A%2F%29\n",
    "onenote:https://microsoft.sharepoint.com/teams/azuremlnursery/SiteAssets/Azure%20ML%20Nursery%20Notebook/Workshop.one#Pipeline%20Scope%20and%20Batch&section-id={265D85D5-44C8-9D40-B556-A31FA098E708}&page-id={E40F0043-079C-4551-BAD6-27E5D056D05A}&end)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "subscription_id =\n",
    "resource_group =\n",
    "workspace_name =\n",
    "workspace_region ="
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Workspace\n",
    "\n",
    "ws = Workspace.from_config()\n",
    "ws = Workspace(subscription_id,resource_group, workspace_name)\n",
    "print(ws.subscription_id, ws.resource_group, ws.name, ws.location, sep = '\\n')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create AML experiment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import datetime\n",
    "date_object = datetime.now()\n",
    "time_format = date_object.strftime('%b%d_%H_%M_')\n",
    "exp_name = time_format + \"Scope_And_ExePool-Exp\"\n",
    "exp = Experiment(ws, exp_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Register the migrated ADLS Datastore\n",
    "For this, you will first need to assign the Azure AD application to the Azure Data Lake Storage Gen1 account file or folder. This is detailed in [this article](https://docs.microsoft.com/en-us/azure/data-lake-store/data-lake-store-service-to-service-authenticate-using-active-directory)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Get the details of the account to be used from this [OneNote page](https://microsoft.sharepoint.com/teams/azuremlnursery/_layouts/OneNote.aspx?id=%2Fteams%2Fazuremlnursery%2FSiteAssets%2FAzure%20ML%20Nursery%20Notebook&wd=target%28Workshop.one%7C265D85D5-44C8-9D40-B556-A31FA098E708%2FPipeline%3A%20Scope%20and%20Batch%7CE40F0043-079C-4551-BAD6-27E5D056D05A%2F%29\n",
    "onenote:https://microsoft.sharepoint.com/teams/azuremlnursery/SiteAssets/Azure%20ML%20Nursery%20Notebook/Workshop.one#Pipeline%20Scope%20and%20Batch&section-id={265D85D5-44C8-9D40-B556-A31FA098E708}&page-id={E40F0043-079C-4551-BAD6-27E5D056D05A}&end)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "adl_datastore_name='MigratedAMLPlayground'\n",
    "# ADLS Details:\n",
    "subscription_id=\n",
    "resource_group=\n",
    "store_name=\n",
    "# Team Service Principal details\n",
    "tenant_id=\n",
    "client_id=\n",
    "client_secret="
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    adls_datastore = Datastore.get(ws, adl_datastore_name)\n",
    "    print(\"Found datastore with name: %s\" % adl_datastore_name)\n",
    "except:\n",
    "    adls_datastore = Datastore.register_azure_data_lake(\n",
    "        workspace=ws,\n",
    "        datastore_name=adl_datastore_name,\n",
    "        subscription_id=subscription_id, # subscription id of ADLS account\n",
    "        resource_group=resource_group, # resource group of ADLS account\n",
    "        store_name=store_name, # ADLS account name\n",
    "        tenant_id=tenant_id, # tenant id of service principal\n",
    "        client_id=client_id, # client id of service principal\n",
    "        client_secret=client_secret) # the secret of service principal\n",
    "    print(\"Registered datastore with name: %s\" % adl_datastore_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define Input and Output for Scope Job\n",
    "Input data is already in the ADLS store. We will write outout data also in the ADLS datastore defined above."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "input_data = DataReference(\n",
    "    datastore=adls_datastore,\n",
    "    data_reference_name=\"InputData\",\n",
    "    path_on_datastore=\"local/AMLTest/input-s.tsv\")\n",
    "\n",
    "output_ref = PipelineData(\"Destination\", datastore=adls_datastore)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create Scope step"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**ScopeStep** is used to run a scope script using cosmos-migrated Azure Data Lake Analytics account.\n",
    "\n",
    "- **name:** Name of module\n",
    "- **script_name:** Name of scope script\n",
    "- **scope_param:** Parameters to pass to scope job\n",
    "- **params:** Dictionary of name-value pairs to replace in script *(optional)*\n",
    "- **custom_job_name_suffix:** Optional string to append to scope job name\n",
    "- **inputs:** List of input port bindings\n",
    "- **outputs:** List of output port bindings\n",
    "- **resources:** List of input port bindings to download resource files and substitute their local path in script\n",
    "- **adla_account_name:** the ADLA account name to use for this job\n",
    "- **source_directory:** folder that contains the script, assemblies etc. *(optional)*\n",
    "- **hash_paths:** list of paths to hash to detect a change (script file is always hashed) *(optional)*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "script_folder = '.'\n",
    "\n",
    "script_step = ScopeStep(\n",
    "    name='Remove_Duplicates',\n",
    "    script_name='script.script',\n",
    "    inputs=[input_data],\n",
    "    outputs=[output_ref],\n",
    "    allow_reuse=True,\n",
    "    adla_account_name='amlplayground-c09', #ADLA Name, could be any ADLA name\n",
    "    source_directory=script_folder)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Copy the output of the ScopeStep to Azure blob\n",
    "### Define data destination\n",
    "#### Register Blob"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    def_blob_store = Datastore(workspace=ws, name=\"myblobdatastore\")\n",
    "    print(\"Got blob\")\n",
    "except:\n",
    "    def_blb_store = Datastore.register_azure_blob_container(\n",
    "        ws, \n",
    "        \"myblobdatastore\", \n",
    "        container_name=\"amltest\", \n",
    "        account_name=\"sanpilinternal\")\n",
    "    print(\"Register the blob\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Define the destination"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "blob_destination = DataReference(datastore=def_blob_store,\n",
    "                       path_on_datastore=\"input\",\n",
    "                       data_reference_name=\"Copy_Destination\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Copy data using DataTransferStep"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Register ADF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core.compute import ComputeTarget, DataFactoryCompute\n",
    "data_factory_name = 'adftest'\n",
    "\n",
    "try:\n",
    "    data_factory_compute = DataFactoryCompute(ws, data_factory_name)\n",
    "    print(\"Got ADF\")\n",
    "except:\n",
    "    print(\"Registering ADF\")\n",
    "    provisioning_config = DataFactoryCompute.provisioning_configuration()\n",
    "    data_factory_compute = ComputeTarget.create(ws, data_factory_name, provisioning_config)\n",
    "    data_factory_compute.wait_for_completion()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Define DataTransferStep"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "transfer_adls_to_blob = DataTransferStep(\n",
    "    name=\"Copy_from_ADLS_to_Blob\",\n",
    "    source_data_reference=output_ref,\n",
    "    destination_data_reference=blob_destination,\n",
    "    source_reference_type='file',\n",
    "    compute_target=data_factory_compute)\n",
    "print(\"data transfer step created\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Run a Windows Exe on Azure Batch\n",
    "Using the output we copied to Azure Blob as the input, run a Windows exe in Azure Batch.\n",
    "### Define AzureBatchTaskInfo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "file_name = \"input/Destination_f99a1b07-252a-43f2-bea1-ea9e3b42e8ed\"\n",
    "batch_input = transfer_adls_to_blob.get_output().as_input('input')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "task_command_line = \"cmd /c wc.exe -w \" + file_name + \" >> output.txt\"\n",
    "\n",
    "blob_out = DataReference(datastore=def_blob_store,\n",
    "                         path_on_datastore=\"\",\n",
    "                         data_reference_name=\"output\")\n",
    "\n",
    "azurebatch_task = AzureBatchTaskInfo(task_name=\"wordcount\",\n",
    "                                     task_command_line=task_command_line,\n",
    "                                     task_output_patterns=[\"output.txt\"],\n",
    "                                     task_input_data_references=[batch_input],\n",
    "                                     task_output_data_reference=blob_out)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "batch_step = AzureBatchStep(\n",
    "            name=\"Word Count\",\n",
    "            account_name=\"batch3pdev\",\n",
    "            pool_id=\"sanpilpool\",\n",
    "            tasks=[azurebatch_task],\n",
    "            source_directory=\"wordcount\"\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Run the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline = Pipeline(\n",
    "    description=\"ScopeAndExePool\",\n",
    "    workspace=ws, \n",
    "#    steps=[script_step])\n",
    "    steps=[script_step, transfer_adls_to_blob, batch_step])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_run = exp.submit(pipeline, regenerate_outputs=False)\n",
    "#pipeline_run.wait_for_completion()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "from azureml.widgets import RunDetails\n",
    "RunDetails(pipeline_run).show()"
   ]
  }
 ],
 "metadata": {
  "authors": [
   {
    "name": "akvasude"
   }
  ],
  "kernelspec": {
   "display_name": "Python [conda env:cli_dev]",
   "language": "python",
   "name": "conda-env-cli_dev-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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}