notebooks/sample_submit_ptm_batch.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load_ext autoreload\n",
    "%autoreload 2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dmsbatch import create_batch_client, create_blob_client\n",
    "import datetime\n",
    "import logging\n",
    "#logger = logging.getLogger()\n",
    "#logger.setLevel(logging.ERROR)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## First create a batch client from the config file\n",
    "\n",
    "The config file is described in the [README](../README.md)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client = create_batch_client('../tests/data/dmsbatch.config')\n",
    "blob_client = create_blob_client('../tests/data/dmsbatch.config')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Application packages\n",
    "To copy large files and programs it is best to zip (or targz) them and upload them as application packages\n",
    "\n",
    "Application packages are setup separately in either azure management apis or from the web console or cli tool\n",
    "\n",
    "These are referenced here by their name and version\n",
    "e.g. DSM2, python and other programs\n",
    "\n",
    "One extra field (last one) is the path within the zip file where the executables can be found. These are used later to setup the PATH varible"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "app_pkgs = [('dsm2', '8.2.c5aacef7', 'DSM2-8.2.c5aacef7-win32/bin'),\n",
    "            ('vista', '1.0-v2019-05-28', 'bin'),\n",
    "            ('unzip', '5.51-1', 'bin')]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Show vms available\n",
    "\n",
    "https://docs.microsoft.com/en-us/azure/virtual-machines/fsv2-series"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#display(client.skus_available())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create or resize existing pool\n",
    "If the pool doesn't exist it will create it\n",
    "If the pool exists, it will resize to the second arg"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pool_name = 'ptmlargepool'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.create_or_resize_pool(pool_name,\n",
    "                   1,\n",
    "                   app_packages=[(app,version) for app,version,_ in app_pkgs], \n",
    "                   vm_size='standard_f32s_v2', \n",
    "                   tasks_per_vm=32,\n",
    "                   os_image_data=('microsoftwindowsserver', 'windowsserver', '2019-datacenter-core')\n",
    "                  )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create job on pool or fail if it exists\n",
    "Jobs are containers of tasks (things that run on nodes (machines) in the pool). If this exists, the next line will fail"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# step is only needed as these files were upload to amazon s3. Otherwise only upload (next step) to Azure is needed\n",
    "# downloading from aws takes 2 - 3 mins\n",
    "#!aws s3 cp s3://ca.dwr.dms.dsm2/release82/ptm_runs/input/DCP_EX_EX.h5 ."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# slow - 9 mins so use max_connections > 2 (default). Using 12 which seems to be a good fit here\n",
    "blob_client.upload_file_to_container('ptmnbjob','DCP_EX.h5','./DCP_EX_EX.h5',max_connections=12)\n",
    "# much faster - 3 mins upload time\n",
    "#azcopy.exe copy \"D:\\dev\\azure_dms_batch\\notebooks\\DCP_EX_EX.h5\" \"https://dwrmodelingstore.blob.core.windows.net/ptmnbjob/DCP_EX_EX.h5?sv=2020-08-04&se=2021-12-26T04%3A22%3A42Z&sr=c&sp=rwl&sig=OCNOFWYMRpJ2lV9w7iAVzAZNttZXX9eDkLqoHyn7qXY%3D\" --overwrite=prompt --from-to=LocalBlob --blob-type Detect --follow-symlinks --put-md5 --follow-symlinks --recursive --trusted-microsoft-suffixes= --log-level=INFO;\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#TODO: client.delete_job"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#input_tidefile = client.create_input_file_spec('ptmnbjob',blob_prefix='DCP_EX.h5',file_path='.')\n",
    "copy_tidefile_task = client.create_task_copy_file_to_shared_dir('ptmnbjob','DCP_EX.h5',file_path='.')\n",
    "client.create_job('ptmnbjob',pool_name,prep_task=copy_tidefile_task)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "local_dir = 'd:/dev/ptm_batch/neutrally_bouyant_particles/ex'\n",
    "input_file=blob_client.zip_and_upload('ptmnbjob',None,local_dir,30)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "input_file = client.create_input_file_spec('ptmnbjob',blob_prefix='ex.zip',file_path='.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dmsbatch\n",
    "permissions = dmsbatch.commands.azureblob.BlobPermissions.WRITE\n",
    "# |helpers.azureblob.BlobPermissions.ADD|helpers.azureblob.BlobPermissions.CREATE\n",
    "output_dir_sas_url = blob_client.get_container_sas_url('ptmnbjob', permissions)\n",
    "print(output_dir_sas_url)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create a task\n",
    "This uses the application package as pre -set up. If not, create one https://docs.microsoft.com/en-us/azure/batch/batch-application-packages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def submit_ptm_single_task(task_name, envvars):\n",
    "    std_out_files = client.create_output_file_spec(\n",
    "        '../std*.txt', output_dir_sas_url, blob_path=f'{task_name}')\n",
    "    output_dir = client.create_output_file_spec(\n",
    "        '**/output/*', output_dir_sas_url, blob_path=f'{task_name}')\n",
    "    cmd_string = client.wrap_cmd_with_app_path(\n",
    "        'unzip ex.zip & del *.zip & set \"TIDEFILE_LOC=%AZ_BATCH_NODE_SHARED_DIR:\\=/%\" & cd studies & ptm.bat planning_ptm.inp & cd output & del trace.out', app_pkgs)\n",
    "    #print(cmd_string)\n",
    "    ptm_task = client.create_task(task_name, cmd_string,\n",
    "                                  resource_files=[input_file],\n",
    "                                  output_files=[\n",
    "                                      std_out_files, output_dir],\n",
    "                                  env_settings=envvars)\n",
    "    return ptm_task"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create all tasks\n",
    "This function looks at the insertion location file and the simulation years and months to create an array of tasks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import csv\n",
    "import logging\n",
    "def create_tasks(insertion_file='run_number_loc.txt',\n",
    "               simulation_start_year=1923,\n",
    "               simulation_end_year=2015,\n",
    "               simulation_start_day=1,\n",
    "               simulation_month=[1, 2, 3, 4, 5, 6],\n",
    "               simulation_days=92,\n",
    "               duration='1485minutes',\n",
    "               delay='0day',\n",
    "               study_name='ex',\n",
    "               study_folder='neutrally_bouyant_particles',\n",
    "               setup_tidefile=False):\n",
    "    tasks = []\n",
    "    with open(insertion_file, 'r') as input:\n",
    "        for row in csv.DictReader(input):  # run#,particle#,node\n",
    "            run_no = row['run#']\n",
    "            particle_no = row['particle#']\n",
    "            insertion_node = row['node']\n",
    "            job_name_prefix = 'ptm-%s-%s-%s' % (\n",
    "                study_folder[0:5], study_name, run_no)\n",
    "            #\n",
    "            sim_days = datetime.timedelta(days=simulation_days)\n",
    "            for y in range(simulation_start_year, simulation_end_year+1):\n",
    "                for m in simulation_month:\n",
    "                    s_day = datetime.date(y, m, simulation_start_day)\n",
    "                    e_day = s_day + sim_days\n",
    "                    ptm_start_date = s_day.strftime(\"%d%b%Y\")\n",
    "                    ptm_end_date = e_day.strftime(\"%d%b%Y\")\n",
    "                    particle_insertion_row = '%s %s %s %s' % (\n",
    "                        insertion_node, particle_no, delay, duration)\n",
    "                    envvars = {'RUN_NO': '%s' % run_no,\n",
    "                               'PTM_START_DATE': '%s' % ptm_start_date,\n",
    "                               'PTM_END_DATE': '%s' % ptm_end_date,\n",
    "                               'PARTICLE_INSERTION_ROW': '%s' % particle_insertion_row,\n",
    "                               'DSM2_STUDY_NAME': 'DCP_%s_%sP' % (study_name, study_folder[0:1])\n",
    "                               }\n",
    "                    task = submit_ptm_single_task(\n",
    "                        job_name_prefix+'-'+ptm_start_date, envvars)\n",
    "                    tasks.append(task)\n",
    "    logging.info('All done!')\n",
    "    return tasks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tasks = create_tasks(insertion_file='d:/dev/ptm_batch/run_number_loc.txt',simulation_start_year=1990,simulation_end_year=1991)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "len(tasks)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tasks[0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Next submit the task and wait \n",
    "Azure batch limits to submitting 100 tasks at a time."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.submit_tasks('ptmnbjob',tasks[0:100])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for i in range(1,round(len(tasks)/100)):\n",
    "    client.submit_tasks('ptmnbjob',tasks[i*100:i*100+100])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Finally resize the pool to 0 to save costs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#client.resize_pool(pool_name,0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "interpreter": {
   "hash": "9e539dc02fc56d55a22d79a0646788fa38cb6ebb3e85aa69306aae5c2f643a8f"
  },
  "kernelspec": {
   "display_name": "Python [conda env:dev_azure]",
   "language": "python",
   "name": "conda-env-dev_azure-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.7.11"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}