Capabilities.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Capabilities Demonstration\n",
    "Shows the features in Jupyter Helper functions. The following are demonstrated below:\n",
    "\n",
    "1. How to connect to spark\n",
    "2. How to specify RUN_ID\n",
    "3. How to init Juyper Helper\n",
    "4. Data Loading\n",
    "    1. How to load business rules/reasons (final scoring)\n",
    "    2. How to load business rules/reasons (scoring engine)\n",
    "    3. How to load final scores\n",
    "    4. How to load rollup\n",
    "    5. How to load rollup participation\n",
    "    6. How to load benchmarks\n",
    "    7. How to load submision scores\n",
    "    8. How to load base scores\n",
    "5. How to load data from UDS"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 1 - Connect to Spark:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Hi J5D6, is your username printed correctly?. No, then run Utilities script.\n"
     ]
    }
   ],
   "source": [
    "%reload_ext sparkmagic.magics\n",
    "import pandas as pd\n",
    "pd.set_option('display.max_colwidth', 100)\n",
    "%store -r myuser\n",
    "%store -r mypass\n",
    "print('Hi ' + myuser + ', is your username printed correctly?. No, then run Utilities script.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "%spark cleanup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>191</td><td>application_1555450382193_0031</td><td>spark</td><td>idle</td><td><a target=\"_blank\" href=\"http://qpp-ar-impl-hdp-0.ar.qpp.internal:8088/proxy/application_1555450382193_0031/\">Link</a></td><td><a target=\"_blank\" href=\"http://qpp-ar-impl-hdp-4.ar.qpp.internal:8042/node/containerlogs/container_e07_1555450382193_0031_01_000001/J5D6\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    }
   ],
   "source": [
    "%spark add -s $myuser -l scala -k -u https://ambari.impl.qppar.internal:8443/qpp-ar-impl-hadoop/default/livy/v1 -a $myuser -p $mypass"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You must wait for the above step to finish. <span style=\"color:red\">some **Do you see the yarn application id ?** No, then keep waiting...</span>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2 - Initialize your Run & Environment"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Which run are you dealing with ?**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "RUN_ID: String = 105"
     ]
    }
   ],
   "source": [
    "%%spark \n",
    "val RUN_ID = \"105\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Initialize Jupyter Helper**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "jupyter: gov.cms.qpp.scoring.jupyter.Jupyter = gov.cms.qpp.scoring.jupyter.Jupyter@354303f6"
     ]
    }
   ],
   "source": [
    "%%spark\n",
    "import gov.cms.qpp.scoring.vo.fs._\n",
    "import gov.cms.qpp.scoring.vo.rollup._\n",
    "import gov.cms.qpp.scoring.vo.bs._\n",
    "import gov.cms.qpp.scoring.vo.benchmarks._\n",
    "import gov.cms.qpp.scoring.jupyter._\n",
    "val jupyter = Jupyter.connect(spark, RUN_ID)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3 - Loading Data (from Hive)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Loading reason codes associated with final scores"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "fsReasons: org.apache.spark.sql.Dataset[gov.cms.qpp.scoring.jupyter.FsReason] = [rule: string, ruleName: string ... 2 more fields]"
     ]
    }
   ],
   "source": [
    "%%spark -o fsReasons\n",
    "val fsReasons = jupyter.loadFsReasons"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fsReasons"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Loading reason codes associated with scoring engine"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "seReasons: org.apache.spark.sql.Dataset[gov.cms.qpp.scoring.jupyter.SeReason] = [id: string, statute: string ... 2 more fields]"
     ]
    }
   ],
   "source": [
    "%%spark -o seReasons\n",
    "val seReasons = jupyter.loadSeReasons"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "seReasons"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load the final score data from spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "fs: org.apache.spark.sql.Dataset[gov.cms.qpp.scoring.vo.fs.FinalScoreVO] = [tin: string, npi: string ... 26 more fields]"
     ]
    }
   ],
   "source": [
    "%%spark -o fs\n",
    "val fs = jupyter.loadFinalScores"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The total number of final score entries in run 105: 1577038"
     ]
    }
   ],
   "source": [
    "%%spark \n",
    "print (s\" The total number of final score entries in run ${RUN_ID}: ${fs.count()}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%spark \n",
    "fs.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load Benchmarks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "benchmarks: org.apache.spark.sql.Dataset[gov.cms.qpp.scoring.vo.benchmarks.MeasureDecilesVO] = [measureId: string, submissionMethod: string ... 2 more fields]"
     ]
    }
   ],
   "source": [
    "%%spark -o benchmarks\n",
    "val benchmarks = jupyter.loadBenchmarks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total benchmarks for 105: 0"
     ]
    }
   ],
   "source": [
    "%%spark \n",
    "print(s\"Total benchmarks for ${RUN_ID}: ${benchmarks.count()}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load the rollup "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "rollup: org.apache.spark.sql.Dataset[gov.cms.qpp.scoring.vo.rollup.PiRollupScoreVO] = [apmEntityId: string, rollupScore: double ... 3 more fields]"
     ]
    }
   ],
   "source": [
    "%%spark -o rollup\n",
    "val rollup = jupyter.loadRollup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total apm entities receiving rollup scores for 105: 2375"
     ]
    }
   ],
   "source": [
    "%%spark\n",
    "println(s\"Total apm entities receiving rollup scores for ${RUN_ID}: ${rollup.count()}\") "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rollup"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load the rollup participation list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "rollupParticipats: org.apache.spark.sql.Dataset[gov.cms.qpp.scoring.vo.rollup.ApmRollupParticipantVO] = [apmEntityId: string, tin: string ... 8 more fields]"
     ]
    }
   ],
   "source": [
    "%%spark -o rollupParticipats\n",
    "val rollupParticipats = jupyter.loadRollupParticipats"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of entries available in rollup participation list for 105: 666958"
     ]
    }
   ],
   "source": [
    "%%spark\n",
    "println(s\"Number of entries available in rollup participation list for ${RUN_ID}: ${rollupParticipats.count()}\") "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rollupParticipats"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load Base scores"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "baseScores: org.apache.spark.sql.Dataset[gov.cms.qpp.scoring.vo.bs.SubmissionScoreVO] = [id: string, tin: string ... 25 more fields]"
     ]
    }
   ],
   "source": [
    "%%spark -o baseScores \n",
    "val baseScores = jupyter.hive(s\"select * from final_scoring.basescore_${RUN_ID}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of entries available in basescores for 105: 187982"
     ]
    }
   ],
   "source": [
    "%%spark\n",
    "println(s\"Number of entries available in basescores for ${RUN_ID}: ${baseScores.count()}\") "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "baseScores"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%spark -o submissionScores \n",
    "val submissionScores = jupyter.hive(s\"select * from final_scoring.submissionscore_${RUN_ID}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load Submission scores"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%spark\n",
    "println(s\"Number of entries available in submission scores for ${RUN_ID}: ${submissionScores.count()}\") "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "submissionScores"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Loading Data (from UDS)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "qps: org.apache.spark.sql.DataFrame = [npi: string, qp_status: string]"
     ]
    }
   ],
   "source": [
    "%%spark -o qps \n",
    "val qps = jupyter.uds(\"\"\"\n",
    "    SELECT DISTINCT npi, qp_status FROM active.provider\n",
    "    WHERE run in (0, 4, 5) AND year = 2018 AND ( qp_status = 'Y' OR qp_status = 'Q' )\n",
    " \"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%spark\n",
    "qps.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- npi: string (nullable = true)\n",
      " |-- qp_status: string (nullable = true)"
     ]
    }
   ],
   "source": [
    "%%spark\n",
    "qps.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Done Now kill spark session "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "%spark cleanup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[0;31mDocstring:\u001b[0m\n",
       "::\n",
       "\n",
       "  %spark [-c CONTEXT] [-s SESSION] [-o OUTPUT] [-q [QUIET]]\n",
       "             [-m SAMPLEMETHOD] [-n MAXROWS] [-r SAMPLEFRACTION] [-u URL]\n",
       "             [-a USER] [-p PASSWORD] [-t AUTH] [-l LANGUAGE] [-k [SKIP]]\n",
       "             [-i ID] [-e COERCE]\n",
       "             [command [command ...]]\n",
       "\n",
       "Magic to execute spark remotely.\n",
       "\n",
       "This magic allows you to create a Livy Scala or Python session against a Livy endpoint. Every session can\n",
       "be used to execute either Spark code or SparkSQL code by executing against the SQL context in the session.\n",
       "When the SQL context is used, the result will be a Pandas dataframe of a sample of the results.\n",
       "\n",
       "If invoked with no subcommand, the cell will be executed against the specified session.\n",
       "\n",
       "Subcommands\n",
       "-----------\n",
       "info\n",
       "    Display the available Livy sessions and other configurations for sessions.\n",
       "add\n",
       "    Add a Livy session given a session name (-s), language (-l), and endpoint credentials.\n",
       "    The -k argument, if present, will skip adding this session if it already exists.\n",
       "    e.g. `%spark add -s test -l python -u https://sparkcluster.net/livy -t Kerberos -a u -p -k`\n",
       "config\n",
       "    Override the livy session properties sent to Livy on session creation. All session creations will\n",
       "    contain these config settings from then on.\n",
       "    Expected value is a JSON key-value string to be sent as part of the Request Body for the POST /sessions\n",
       "    endpoint in Livy.\n",
       "    e.g. `%%spark config`\n",
       "         `{\"driverMemory\":\"1000M\", \"executorCores\":4}`\n",
       "run\n",
       "    Run Spark code against a session.\n",
       "    e.g. `%%spark -s testsession` will execute the cell code against the testsession previously created\n",
       "    e.g. `%%spark -s testsession -c sql` will execute the SQL code against the testsession previously created\n",
       "    e.g. `%%spark -s testsession -c sql -o my_var` will execute the SQL code against the testsession\n",
       "             previously created and store the pandas dataframe created in the my_var variable in the\n",
       "             Python environment.\n",
       "logs\n",
       "    Returns the logs for a given session.\n",
       "    e.g. `%spark logs -s testsession` will return the logs for the testsession previously created\n",
       "delete\n",
       "    Delete a Livy session.\n",
       "    e.g. `%spark delete -s defaultlivy`\n",
       "cleanup\n",
       "    Delete all Livy sessions created by the notebook. No arguments required.\n",
       "    e.g. `%spark cleanup`\n",
       "\n",
       "positional arguments:\n",
       "  command               Commands to execute.\n",
       "\n",
       "optional arguments:\n",
       "  -c CONTEXT, --context CONTEXT\n",
       "                        Context to use: 'spark' for spark and 'sql' for sql\n",
       "                        queries. Default is 'spark'.\n",
       "  -s SESSION, --session SESSION\n",
       "                        The name of the Livy session to use.\n",
       "  -o OUTPUT, --output OUTPUT\n",
       "                        If present, output when using SQL queries will be\n",
       "                        stored in this variable.\n",
       "  -q <[QUIET]>, --quiet <[QUIET]>\n",
       "                        Do not display visualizations on SQL queries\n",
       "  -m SAMPLEMETHOD, --samplemethod SAMPLEMETHOD\n",
       "                        Sample method for SQL queries: either take or sample\n",
       "  -n MAXROWS, --maxrows MAXROWS\n",
       "                        Maximum number of rows that will be pulled back from\n",
       "                        the server for SQL queries\n",
       "  -r SAMPLEFRACTION, --samplefraction SAMPLEFRACTION\n",
       "                        Sample fraction for sampling from SQL queries\n",
       "  -u URL, --url URL     URL for Livy endpoint\n",
       "  -a USER, --user USER  Username for HTTP access to Livy endpoint\n",
       "  -p PASSWORD, --password PASSWORD\n",
       "                        Password for HTTP access to Livy endpoint\n",
       "  -t AUTH, --auth AUTH  Auth type for HTTP access to Livy endpoint. [Kerberos,\n",
       "                        None, Basic Auth]\n",
       "  -l LANGUAGE, --language LANGUAGE\n",
       "                        Language for Livy session; one of python, scala, r\n",
       "  -k <[SKIP]>, --skip <[SKIP]>\n",
       "                        Skip adding session if it already exists\n",
       "  -i ID, --id ID        Session ID\n",
       "  -e COERCE, --coerce COERCE\n",
       "                        Whether to automatically coerce the types (default,\n",
       "                        pass True if being explicit) of the dataframe or not\n",
       "                        (pass False)\n",
       "\u001b[0;31mFile:\u001b[0m      /opt/conda/lib/python3.6/site-packages/sparkmagic/livyclientlib/exceptions.py\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%spark?"
   ]
  },
  {
   "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.6.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}