streaming-example.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import sys\n",
    "from pyspark import SparkContext, SparkConf\n",
    "from streaming.encounter_job import EncounterJob\n",
    "from common.utils import PipelineUtils\n",
    "spark=PipelineUtils.getSpark()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "job = EncounterJob()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Encounter Streaming Job Started at = 2020-03-18 09:28:28.343346\n",
      "Event received in window:  None\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:30\n",
      "-------------------------------------------\n",
      "Row(encounters=[112, 113, 114, 115, 116, 105, 106, 107, 108, 109, 110, 111], person_id=16)\n",
      "Row(encounters=[528, 529, 530, 531, 532, 533, 534, 535, 536, 527], person_id=48)\n",
      "Row(encounters=[720, 721, 722, 723, 724, 725, 726, 727, 728, 715, 731, 716, 732, 717, 718, 719], person_id=64)\n",
      "Row(encounters=[912, 903, 904, 905, 906, 907, 908, 909, 910, 911], person_id=80)\n",
      "Row(encounters=[1104, 1105, 1106, 1107, 1108, 1101, 1102, 1103], person_id=96)\n",
      "Row(encounters=[1264, 1265, 1266, 1263], person_id=112)\n",
      "Row(encounters=[1600, 1616, 1601, 1602, 1603, 1619, 1604, 1620, 1605, 1606, 1607, 1608, 1609, 1610, 1611, 1612, 1613, 1614, 1599, 1615], person_id=144)\n",
      "Row(encounters=[1744, 1760, 1745, 1746, 1747, 1748, 1749, 1750, 1751, 1752, 1753, 1754, 1755, 1756, 1743, 1759], person_id=160)\n",
      "Row(encounters=[2352, 2353, 2354, 2355, 2356, 2345, 2346, 2347, 2348, 2349, 2350, 2351], person_id=224)\n",
      "Row(encounters=[2560, 2561, 2562, 2563, 2564, 2553, 2554, 2555, 2556, 2557, 2558, 2559], person_id=240)\n",
      "...\n",
      "\n",
      "---Micro-Batch--- \n",
      "Starting calculations for flat_enc_obs_orders Wed Mar 18 12:28:53 2020\n",
      "CDC: # Patient IDs in Microbatch -->  453\n",
      "CDC: # Encounter IDs in Microbatch -->  4764\n",
      "Upserting MicroBatch --->  4764\n",
      "MicroBatch Logs:  None\n",
      "Took 44.03351 seconds\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:32\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:34\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:36\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:38\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:40\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:42\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:44\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:46\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:48\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:50\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:52\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:54\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:56\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:28:58\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:00\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:02\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:04\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:06\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:08\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:10\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:12\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:14\n",
      "-------------------------------------------\n",
      "Row(encounters=[5623], person_id=392)\n",
      "\n",
      "---Micro-Batch--- \n",
      "Starting calculations for flat_enc_obs_orders Wed Mar 18 12:29:48 2020\n",
      "CDC: # Patient IDs in Microbatch -->  1\n",
      "CDC: # Encounter IDs in Microbatch -->  1\n",
      "Upserting MicroBatch --->  1\n",
      "MicroBatch Logs:  None\n",
      "Took 15.020098 seconds\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:16\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:18\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:20\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:22\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:24\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:26\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:28\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:30\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:32\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:34\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:36\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:38\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:40\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:42\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:44\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:46\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:48\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:50\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:52\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:54\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:56\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:29:58\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:00\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:02\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:04\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:06\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:08\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:10\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:12\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:14\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:16\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:18\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:20\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:22\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:24\n",
      "-------------------------------------------\n",
      "\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:26\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:28\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:30\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:32\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:34\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:36\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:38\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:40\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:42\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:44\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:46\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:48\n",
      "-------------------------------------------\n",
      "Row(encounters=[5623], person_id=392)\n",
      "\n",
      "---Micro-Batch--- \n",
      "Starting calculations for flat_enc_obs_orders Wed Mar 18 12:30:48 2020\n",
      "CDC: # Patient IDs in Microbatch -->  1\n",
      "CDC: # Encounter IDs in Microbatch -->  1\n",
      "Upserting MicroBatch --->  1\n",
      "MicroBatch Logs:  None\n",
      "Took 16.691358 seconds\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:50\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:52\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:54\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:56\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:30:58\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:00\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:02\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:04\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:06\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:08\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:10\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:12\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:14\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:16\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:18\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:20\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:22\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:24\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:26\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:28\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:30\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:32\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:34\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:36\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:38\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:40\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:42\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:44\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2020-03-18 12:31:46\n",
      "-------------------------------------------\n",
      "\n"
     ]
    }
   ],
   "source": [
    "job.run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check the status: http://localhost:4040/streaming/"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.7.4 64-bit ('anaconda3': virtualenv)",
   "language": "python",
   "name": "python37464bitanaconda3virtualenvefdffc192b144e2b98dfc27892fa181f"
  },
  "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.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}