parallelrunstep_hang_model_name_MOD.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is a modified version of https://github.com/Azure/MachineLearningNotebooks/blob/master/how-to-use-azureml/machine-learning-pipelines/parallel-run/tabular-dataset-inference-iris.ipynb\n",
    "Valerie Carey modified this script to demonstrate a serious Azure bug involving models with vs. without datasets.\n",
    "\n",
    "Original Link Copyright (c) Microsoft Corporation. All rights reserved.\n",
    "Licensed under the MIT License.\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h1> Parallel Run Azure Bug With Model Including a Dataset - Summary </h1>\n",
    "\n",
    "Modified version of https://github.com/Azure/MachineLearningNotebooks/blob/master/how-to-use-azureml/machine-learning-pipelines/parallel-run/tabular-dataset-inference-iris.ipynb\n",
    "\n",
    "I am finding that a ParallelRunStep hangs forever when you pass the name of a registered model that contains any dataset(s) in an argument called \"--model_name\".  The parallel run appears to fail to schedule any mini batches and hangs wihtout scheduling.  The error occurs whether or not you ever use the model in the script.  Here I show a dummy script that contains arguments and does nothing.  The script passes if \"--model_name\" corresponds to a registered model without datasets, but will hang forever for the same model with a dataset.\n",
    "\n",
    "In addition, the argument \"--model_name\" seems to be reserved even though that is not documented anywhere as far as I can tell.  If you pass some dummy string to the argument called \"--model_name\", the script will create an error.  This is true even if that argument isn't actually used anywhere in the script, or if you didn't intend it to refer to a registered model.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h4> Prerequisites </h4>\n",
    "\n",
    "I have deleted the parts of the example code that connect to a workspace, get blob storage, create a compute instance, etc.  I assume you have these resources already.  You can modify the code below to connect to your own resources"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2> Connect to workspace </h2>\n",
    "Modify the code below to connect to your own workspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Check core SDK version number. I had 1.20.0\n",
    "import azureml.core\n",
    "\n",
    "print(\"SDK version:\", azureml.core.VERSION)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Enter your info here!\n",
    "WORKSPACE_NAME = 'YOUR-INFO-HERE'\n",
    "WORKSPACE_SUBSCRIPTION_ID = \"YOUR-INFO-HERE\"\n",
    "WORKSPACE_RESOURCE_GROUP = \"YOUR-INFO-HERE\"\n",
    "\n",
    "from azureml.core.workspace import Workspace\n",
    "\n",
    "ws = Workspace.get(name=WORKSPACE_NAME,\n",
    "               subscription_id=WORKSPACE_SUBSCRIPTION_ID,\n",
    "               resource_group=WORKSPACE_RESOURCE_GROUP)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2> Set your datastore and compute </h2>\n",
    "Modify the code below to specify your own datastore and blob storage locations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Datastore\n",
    "\n",
    "# input datastore\n",
    "iris_data = Datastore.get(ws, 'YOUR-INFO-HERE')\n",
    "\n",
    "# path on the datastore where you will store the model object and data files\n",
    "iris_data_path = 'YOUR/INFO/HERE'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Name of the compute target which should be provisioned already, and have at least 2 nodes\n",
    "COMPUTE_TARGET_NAME = 'YOUR-INFO-HERE'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2> Set output folder </h2>\n",
    "I use the same datastore for outputs as inputs.  You could put in your own information"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.pipeline.core import PipelineData\n",
    "output_folder = PipelineData(name='inferences', datastore=iris_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2> Get the data </h2>\n",
    "Use the \"iris\" dataset as a simple input. This data isn't \"used\" here, except to enable a parallel run to happen (I batch on this data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get a temporary location for storage of downloaded items\n",
    "import tempfile\n",
    "iris_data_tmpdir = tempfile.mkdtemp()\n",
    "print(iris_data_tmpdir)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get the iris dataset, as CSV and Parquet\n",
    "\n",
    "from sklearn import datasets\n",
    "\n",
    "iris = datasets.load_iris()\n",
    "\n",
    "import pandas as pd\n",
    "iris_df = pd.DataFrame(data=iris['data'], columns = iris['feature_names'])\n",
    "print(len(iris_df))\n",
    "\n",
    "# Save as CSV\n",
    "iris_data_local_csv =  os.path.join(iris_data_tmpdir, 'iris.csv')\n",
    "iris_df.to_csv(iris_data_local_csv, sep = ',', index = False)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# list temporary directory contents\n",
    "os.listdir(iris_data_tmpdir)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h4> Create tablular dataset for batching </h4>\n",
    "Move the iris data to blob storate and create dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Move files to blob storage\n",
    "iris_data.upload_files(files=[iris_data_local_csv],\n",
    "                              target_path=iris_data_path, overwrite=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create dataset\n",
    "from azureml.core.dataset import Dataset\n",
    "\n",
    "iris_ds_csv = Dataset.Tabular.from_delimited_files(path=[(iris_data,  \n",
    "                                                           '/'.join([iris_data_path, 'iris.csv']))], \n",
    "                                                           validate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2> Register Models </h2>\n",
    "Note that models are used nowhere in the code.  However, how and whether a model is registered affects whether a simple test script runs. Therefore I register the iris model a couple different ways.\n",
    "I get data from the sklearn iris dataset\n",
    "https://github.com/Azure-Samples/Machine-Learning-Operationalization/blob/master/samples/python/code/iris/model.pkl"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get the model object from a Github repo\n",
    "\n",
    "import requests \n",
    "model_url = 'https://github.com/Azure-Samples/Machine-Learning-Operationalization/raw/master/samples/python/code/iris/model.pkl'\n",
    "r = requests.get(model_url, allow_redirects=True)\n",
    "model_data_local = os.path.join(iris_data_tmpdir, 'model.pkl')\n",
    "\n",
    "open(model_data_local, 'wb').write(r.content)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h4> Register usual model </h4>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core.model import Model\n",
    "\n",
    "model1 = Model.register(model_path = model_data_local,\n",
    "                       model_name = \"iris-prs\", \n",
    "                       tags = {'pretrained': \"iris\"},\n",
    "                       workspace = ws)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h4> Register model plus a dataset </h4>\n",
    "I register the same model, but add a dataset.  I use the same dataset as above although you could add any dataset or more than 1.  Somehow adding a dataset makes the parallel run step fail, even if a model isn't used ¯\\_(ツ)_/¯"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model2 = Model.register(model_path = model_data_local,\n",
    "                       model_name = \"iris-prs2\", \n",
    "                        datasets = [('data', iris_ds_csv)],\n",
    "                       tags = {'pretrained': \"iris\"},\n",
    "                       workspace = ws)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2> Create experiment script </h2>\n",
    "I use this notebook to output a .py file. This doesn't actually do anything with the input data.  The purpose is to show that it finishes with certain arguments, hangs forever with others"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get a temporary location for storage of downloaded items\n",
    "import tempfile\n",
    "py_tmpdir = tempfile.mkdtemp()\n",
    "print(py_tmpdir)\n",
    "\n",
    "py_outfile = os.path.join(py_tmpdir, 'iris_score.py')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile $py_outfile\n",
    "import io\n",
    "import argparse\n",
    "import pandas as pd\n",
    "\n",
    "from azureml_user.parallel_run import EntryScript\n",
    "\n",
    "def init():\n",
    "\n",
    "    logger = EntryScript().logger\n",
    "    logger.info(\"init() is called.\")\n",
    "\n",
    "    # Define 2 input parameters to show special behavior of \"model_name\" string\n",
    "    parser = argparse.ArgumentParser(description=\"Iris model serving\")\n",
    "    parser.add_argument('--arg_name', dest=\"arg_name\", required=False)\n",
    "    parser.add_argument('--model_name', dest=\"arg_name\", required=False)\n",
    "    args, unknown_args = parser.parse_known_args()\n",
    "\n",
    "\n",
    "def run(input_data):\n",
    "    # Return nonsense data frame\n",
    "    result=pd.DataFrame({'A' : [1]})\n",
    "    return result\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "os.listdir(py_tmpdir)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2> Set up for batch run </h2>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Set the environment\n",
    "from azureml.core import Environment\n",
    "from azureml.core.runconfig import CondaDependencies\n",
    "\n",
    "predict_conda_deps = CondaDependencies.create(pip_packages=[\"scikit-learn==0.20.3\",\n",
    "                                                            \"azureml-core\", \"azureml-dataset-runtime[pandas,fuse]\"])\n",
    "\n",
    "predict_env = Environment(name=\"predict_environment\")\n",
    "predict_env.python.conda_dependencies = predict_conda_deps\n",
    "predict_env.docker.enabled = True\n",
    "predict_env.spark.precache_packages = False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Configure the parallel run\n",
    "\n",
    "from azureml.pipeline.steps import ParallelRunStep, ParallelRunConfig\n",
    "\n",
    "parallel_run_config = ParallelRunConfig(\n",
    "    source_directory=py_tmpdir,\n",
    "    entry_script='iris_score.py',  # the user script to run against each input\n",
    "    mini_batch_size='1KB',\n",
    "    error_threshold=-1,       # Don't worry about errors as I return dummy data\n",
    "    output_action='append_row',\n",
    "    append_row_file_name=\"iris_outputs.txt\",\n",
    "    environment=predict_env,\n",
    "    compute_target=COMPUTE_TARGET_NAME, \n",
    "    node_count=2,\n",
    "    run_invocation_timeout=600\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2> Test 1 - Passing Case with Simple Registered Model </h2>\n",
    "I pass the name of a model which contains no datasets to the script. Note that this works fine, and uses 2-3 mini batches/processes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the pipeline step\n",
    "distributed_csv_iris_step = ParallelRunStep(\n",
    "    name='example-iris-csv',\n",
    "    inputs=[iris_ds_csv.as_named_input('iris_data')],\n",
    "    output=output_folder,\n",
    "    parallel_run_config=parallel_run_config,\n",
    "    arguments=['--model_name', 'iris-prs'], # Passing simple model's name\n",
    "    allow_reuse=False\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run the pipeline\n",
    "from azureml.core import Experiment\n",
    "from azureml.pipeline.core import Pipeline\n",
    "\n",
    "pipeline = Pipeline(workspace=ws, steps=[distributed_csv_iris_step])\n",
    "\n",
    "pipeline_run = Experiment(ws, 'iris-prs').submit(pipeline)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "## Wait the run for completion \n",
    "pipeline_run.wait_for_completion(show_output=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At this point you can view the experiment and it should be successful.  Note that more than 1 mini batches are used!!  I get 3 mini batches.  The script does nothing, which is fine.  The important thing is that it finishes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2> Test 2 - Hanging Case for Model Containing Datasets </h2>\n",
    "The parallel run step will hang forever, when run with a model containing a dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the pipeline step\n",
    "distributed_csv_iris_step2 = ParallelRunStep(\n",
    "    name='example-iris-csv',\n",
    "    inputs=[iris_ds_csv.as_named_input('iris_data')],\n",
    "    output=output_folder,\n",
    "    parallel_run_config=parallel_run_config,\n",
    "    arguments=['--model_name', 'iris-prs2'],   # Using model with dataset's name\n",
    "    allow_reuse=False\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline2 = Pipeline(workspace=ws, steps=[distributed_csv_iris_step2])\n",
    "\n",
    "pipeline_run2 = Experiment(ws, 'iris-prs').submit(pipeline2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Wait the run for completion \n",
    "pipeline_run2.wait_for_completion(show_output=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This script never finishes.  If you look at the step's logs, there is a file called log/job_progress_overview.XXXXXXXX.txt (the XXXXXXXX's are digits- appears to be a date).  If you open that log, you will see something like:\n",
    "<pre>\n",
    "  2021-03-19T14:39:25.891499 Start the simulator.\n",
    "  2021-03-19T14:39:26.798246 The overviewer on 10.0.0.7 started.\n",
    "  2021-03-19T14:39:57.211507 Scheduled 0 mini batches in 31 seconds.\n",
    "  2021-03-19T14:39:57.311094 Processed 0 mini batches in 0:00:31.\n",
    "  2021-03-19T14:40:07.444707 Scheduled 0 mini batches in 42 seconds.\n",
    "  2021-03-19T14:40:07.536334 Processed 0 mini batches in 0:00:42.\n",
    "  ...\n",
    "  2021-03-19T14:47:59.784273 Scheduled 0 mini batches in 514 seconds.\n",
    "  2021-03-19T14:47:59.905612 Processed 0 mini batches in 0:08:34.\n",
    "  2021-03-19T14:48:10.040731 Scheduled 0 mini batches in 524 seconds.\n",
    "  2021-03-19T14:48:10.180070 Processed 0 mini batches in 0:08:44.\n",
    "  </pre>\n",
    "From the above, and comparing to logs from successful runs, it appears that the scheduling never happens.  The error occurs around this step - but you get no \"error\" just a hang.  \"User\" logs are created and you can see the entry_script_log's, which show that the init() function in the script was called."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2> Test 3: Passing Case for Model Name Argument Not Called model_name </h2>\n",
    "The \"--model-name\" argument seems to be some sort of reserved word - the long name works if you rename the argument from \"--model_name\" to \"--arg_name\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the pipeline step\n",
    "distributed_csv_iris_step3 = ParallelRunStep(\n",
    "    name='example-iris-csv',\n",
    "    inputs=[iris_ds_csv.as_named_input('iris_data')],\n",
    "    output=output_folder,\n",
    "    parallel_run_config=parallel_run_config,\n",
    "    arguments=['--arg_name', 'iris-prs2'],   # Using model with dataset's name again\n",
    "    allow_reuse=False\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline3 = Pipeline(workspace=ws, steps=[distributed_csv_iris_step3])\n",
    "\n",
    "pipeline_run3 = Experiment(ws, 'iris-prs').submit(pipeline3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Wait the run for completion \n",
    "pipeline_run3.wait_for_completion(show_output=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This seems fine.  \"model_name\" is a reserved argument name?  That's news to me! I tried to find where that is documented and failed"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2> Test 4: Error Case when model_name Not a Registered Model </h2>\n",
    "If you pass in an arbitrary string for the \"--model-name\" argument, the script will error out after a long wait.  This is true even for my test script where the argument is unused!  The user needs to be aware that model_name is a reserved argument and must refer to an actual model!  The time it takes to error out is long enough that I thought this case was also a hang for a while. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the pipeline step\n",
    "distributed_csv_iris_step4 = ParallelRunStep(\n",
    "    name='example-iris-csv',\n",
    "    inputs=[iris_ds_csv.as_named_input('iris_data')],\n",
    "    output=output_folder,\n",
    "    parallel_run_config=parallel_run_config,\n",
    "    arguments=['--model_name', 'this_is_not_a_real_model'],   # Pass a string that's not a model name\n",
    "    allow_reuse=False\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline4 = Pipeline(workspace=ws, steps=[distributed_csv_iris_step4])\n",
    "\n",
    "pipeline_run4 = Experiment(ws, 'iris-prs').submit(pipeline4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Wait the run for completion \n",
    "pipeline_run4.wait_for_completion(show_output=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At this point you get another hang, but you see slightly different behavior.  The \"user\" entry script logs aren't created -- it appears no part of the script is called. If you look in logs/sys/node_launcher/error.txt, you see:\n",
    "<pre>\n",
    "  {\n",
    "    \"error\": {\n",
    "        \"message\": \"ModelNotFound: Model with name this_is_not_a_real_model not found in provided workspace\"\n",
    "        }\n",
    "  }\n",
    "</pre>\n",
    "This shows the process is looking for a model - the model_name argument is reserved and assumptions are made about its meaning.   Note for the \"model with dataset\" hang (test 2), there is no error file in logs/sys/node_launcher."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "authors": [
   {
    "name": "joringer"
   },
   {
    "name": "asraniwa"
   },
   {
    "name": "pansav"
   },
   {
    "name": "tracych"
   }
  ],
  "categories": [
   "how-to-use-azureml",
   "machine-learning-pipelines",
   "parallel-run"
  ],
  "category": "Other notebooks",
  "compute": [
   "AML Compute"
  ],
  "datasets": [
   "IRIS"
  ],
  "deployment": [
   "None"
  ],
  "exclude_from_index": false,
  "framework": [
   "None"
  ],
  "friendly_name": "IRIS data inferencing using ParallelRunStep",
  "index_order": 1,
  "kernelspec": {
   "display_name": "Python 3.6 - AzureML",
   "language": "python",
   "name": "python3-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"
  },
  "tags": [
   "Batch Inferencing",
   "Pipeline"
  ],
  "task": "Recognize flower type"
 },
 "nbformat": 4,
 "nbformat_minor": 4
}