describe_dask/describe_dask.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuclio: ignore\n",
    "import nuclio"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%nuclio config kind = \"job\"\n",
    "%nuclio config spec.image = \"mlrun/ml-models\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import warnings\n",
    "warnings.simplefilter(action=\"ignore\", category=FutureWarning)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import matplotlib.pyplot as plt\n",
    "import seaborn as sns\n",
    "\n",
    "import dask\n",
    "import dask.dataframe as dd\n",
    "from dask.distributed import Client\n",
    "\n",
    "from mlrun.execution import MLClientCtx\n",
    "from mlrun.datastore import DataItem\n",
    "from mlrun.artifacts import PlotArtifact, TableArtifact\n",
    "from mlrun.mlutils import gcf_clear\n",
    "\n",
    "from yellowbrick import ClassBalance\n",
    "\n",
    "from typing import List"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pd.set_option(\"display.float_format\", lambda x: \"%.2f\" % x)\n",
    "\n",
    "def summarize(\n",
    "    context: MLClientCtx,\n",
    "    dask_key: str = \"dask_key\",\n",
    "    label_column: str = \"labels\",\n",
    "    class_labels: List[str] = [],\n",
    "    plot_hist: bool = True,\n",
    "    plots_dest: str = \"plots\",\n",
    "    alt_scheduler: str = None\n",
    ") -> None:\n",
    "    \"\"\"Summarize a table\n",
    "    \n",
    "    Connects to dask client through the function context, or through an optional\n",
    "    user-supplied scheduler.\n",
    "\n",
    "    :param context:         the function context\n",
    "    :param dask_key:        key of dataframe in dask client \"datasets\" attribute\n",
    "    :param label_column:    ground truth column label\n",
    "    :param class_labels:    label for each class in tables and plots\n",
    "    :param plot_hist:       (True) set this to False for large tables\n",
    "    :param plots_dest:      destination folder of summary plots (relative to artifact_path)\n",
    "    :param alt_scheduler:   (None) an alternative scheduler file to connect with\n",
    "    \"\"\"\n",
    "    if alt_scheduler:\n",
    "        dask_client = Client(scheduler_file=str(alt_scheduler))\n",
    "    elif hasattr(context, \"dask_client\"):\n",
    "        dask_client = Client(scheduler_file=str(context.dask_client))\n",
    "    else:\n",
    "        raise Exception(\"out of luck, no dask_client or scheduler file!\")\n",
    "        \n",
    "    if dask_key in dask_client.datasets:\n",
    "        table = dask_client.get_dataset(dask_key)\n",
    "    else:\n",
    "        context.logger.info(f\"only these datasets are available {dask_client.datasets} in client {dask_client}\")\n",
    "        raise Exception(\"dataset not found on dask cluster\")\n",
    "    header = table.columns.values\n",
    "    \n",
    "    gcf_clear(plt)\n",
    "    table = table.compute()\n",
    "    snsplt = sns.pairplot(table, hue=label_column, diag_kws={'bw': 1.5})\n",
    "    context.log_artifact(PlotArtifact('histograms',  body=plt.gcf()), \n",
    "                         local_path=f\"{plots_dest}/hist.html\")\n",
    "\n",
    "    gcf_clear(plt)   \n",
    "    labels = table.pop(label_column)\n",
    "    if not class_labels:\n",
    "        class_labels = labels.unique()\n",
    "    class_balance_model = ClassBalance(labels=class_labels)\n",
    "    class_balance_model.fit(labels)   \n",
    "    scale_pos_weight = class_balance_model.support_[0]/class_balance_model.support_[1]\n",
    "    context.log_result(\"scale_pos_weight\", f\"{scale_pos_weight:0.2f}\")\n",
    "    context.log_artifact(PlotArtifact(\"imbalance\", body=plt.gcf()), \n",
    "                         local_path=f\"{plots_dest}/imbalance.html\")\n",
    "    \n",
    "    gcf_clear(plt)\n",
    "    tblcorr = table.corr()\n",
    "    ax = plt.axes()\n",
    "    sns.heatmap(tblcorr, ax=ax, annot=False, cmap=plt.cm.Reds)\n",
    "    ax.set_title(\"features correlation\")\n",
    "    context.log_artifact(PlotArtifact(\"correlation\",  body=plt.gcf()), \n",
    "                         local_path=f\"{plots_dest}/corr.html\")\n",
    "    # otherwise shows last plot:\n",
    "    gcf_clear(plt)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuclio: end-code"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### mlconfig"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from mlrun import mlconf\n",
    "import os\n",
    "\n",
    "mlconf.dbpath = mlconf.dbpath or 'http://mlrun-api:8080'\n",
    "mlconf.artifact_path = mlconf.artifact_path or f'{os.environ[\"HOME\"]}/artifacts'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### save"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from mlrun import code_to_function \n",
    "# create job function object from notebook code\n",
    "fn = code_to_function('describe_dask')\n",
    "\n",
    "# add metadata (for templates and reuse)\n",
    "fn.spec.default_handler = 'summarize'\n",
    "fn.spec.description = \"describe and visualizes dataset stats\"\n",
    "fn.metadata.categories = [\"analysis\"]\n",
    "fn.metadata.labels = {'author': 'yjb'}\n",
    "fn.export('function.yaml')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## tests"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# load function from marketplacen\n",
    "from mlrun import import_function\n",
    "\n",
    "# vcs_branch = 'development'\n",
    "# base_vcs = f'https://raw.githubusercontent.com/mlrun/functions/{vcs_branch}/'\n",
    "# mlconf.hub_url = mlconf.hub_url or base_vcs + f'{name}/function.yaml'\n",
    "# fn = import_function(\"hub://describe_dask\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from mlrun import import_function, NewTask, run_local\n",
    "\n",
    "if \"V3IO_HOME\" in list(os.environ):\n",
    "    from mlrun import mount_v3io\n",
    "    fn.apply(mount_v3io())\n",
    "else:\n",
    "    # is you set up mlrun using the instructions at https://github.com/mlrun/mlrun/blob/master/hack/local/README.md\n",
    "    from mlrun.platforms import mount_pvc\n",
    "    fn.apply(mount_pvc('nfsvol', 'nfsvol', '/home/joyan/data'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "task = NewTask(name=\"tasks describe dask\", \n",
    "               inputs={'dask_key': \"dask_key\",\n",
    "                       \"alt_scheduler\" :\"/User/artifacts/scheduler.json\"})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "run = fn.run(task)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}