Monitor Delta Tables.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()\n",
    "#print(SparkConf().getAll())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>encounter_id</th><th>patient_id</th><th>location_id</th><th>visit_id</th><th>encounter_datetime</th><th>encounter_type</th><th>dead</th><th>gender</th><th>death_date</th><th>patient_uuid</th><th>visit_type_id</th><th>birthdate</th><th>obs</th><th>orders</th></tr>\n",
       "<tr><td>5641</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 03:55:28</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:430530...</td><td>null</td></tr>\n",
       "<tr><td>5640</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 03:55:20</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:430528...</td><td>null</td></tr>\n",
       "<tr><td>5637</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 03:48:49</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:430522...</td><td>null</td></tr>\n",
       "<tr><td>5636</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 03:48:34</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:430520...</td><td>null</td></tr>\n",
       "<tr><td>5635</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 03:32:10</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:430518...</td><td>null</td></tr>\n",
       "<tr><td>5634</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 03:31:55</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:430516...</td><td>null</td></tr>\n",
       "<tr><td>5633</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 03:31:32</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:430514...</td><td>null</td></tr>\n",
       "<tr><td>5632</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 00:37:12</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:430507...</td><td>null</td></tr>\n",
       "<tr><td>5631</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 00:25:34</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:430500...</td><td>null</td></tr>\n",
       "<tr><td>5630</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 00:22:29</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:430493...</td><td>null</td></tr>\n",
       "<tr><td>5629</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 00:20:24</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:430486...</td><td>null</td></tr>\n",
       "<tr><td>5628</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 00:16:29</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:430479...</td><td>null</td></tr>\n",
       "<tr><td>5627</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 00:14:58</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:430471...</td><td>null</td></tr>\n",
       "<tr><td>5626</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 00:05:57</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:422275...</td><td>null</td></tr>\n",
       "<tr><td>5625</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 00:05:22</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:420220...</td><td>null</td></tr>\n",
       "<tr><td>5624</td><td>392</td><td>4</td><td>2383</td><td>2020-03-22 00:04:07</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:416118...</td><td>null</td></tr>\n",
       "<tr><td>5623</td><td>392</td><td>4</td><td>2383</td><td>2020-03-21 23:54:21</td><td>2</td><td>false</td><td>M</td><td>null</td><td>90d81642-6276-483...</td><td>1</td><td>1961-10-01</td><td>[{&quot;obs_id&quot;:387452...</td><td>null</td></tr>\n",
       "<tr><td>4336</td><td>394</td><td>1</td><td>1836</td><td>2017-01-18 10:50:03</td><td>5</td><td>false</td><td>M</td><td>null</td><td>c0cbe231-deb8-4cf...</td><td>1</td><td>1972-12-07</td><td>[{&quot;obs_id&quot;:35662,...</td><td>null</td></tr>\n",
       "<tr><td>4335</td><td>394</td><td>7</td><td>1836</td><td>2017-01-18 10:01:03</td><td>2</td><td>false</td><td>M</td><td>null</td><td>c0cbe231-deb8-4cf...</td><td>1</td><td>1972-12-07</td><td>[{&quot;obs_id&quot;:35654,...</td><td>null</td></tr>\n",
       "<tr><td>4950</td><td>444</td><td>1</td><td>2094</td><td>2017-01-17 11:09:58</td><td>5</td><td>false</td><td>M</td><td>null</td><td>a793f92a-87d3-4c4...</td><td>1</td><td>1956-04-07</td><td>[{&quot;obs_id&quot;:40700,...</td><td>null</td></tr>\n",
       "</table>\n",
       "only showing top 20 rows\n"
      ],
      "text/plain": [
       "+------------+----------+-----------+--------+-------------------+--------------+-----+------+----------+--------------------+-------------+----------+--------------------+------+\n",
       "|encounter_id|patient_id|location_id|visit_id| encounter_datetime|encounter_type| dead|gender|death_date|        patient_uuid|visit_type_id| birthdate|                 obs|orders|\n",
       "+------------+----------+-----------+--------+-------------------+--------------+-----+------+----------+--------------------+-------------+----------+--------------------+------+\n",
       "|        5641|       392|          4|    2383|2020-03-22 03:55:28|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":430530...|  null|\n",
       "|        5640|       392|          4|    2383|2020-03-22 03:55:20|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":430528...|  null|\n",
       "|        5637|       392|          4|    2383|2020-03-22 03:48:49|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":430522...|  null|\n",
       "|        5636|       392|          4|    2383|2020-03-22 03:48:34|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":430520...|  null|\n",
       "|        5635|       392|          4|    2383|2020-03-22 03:32:10|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":430518...|  null|\n",
       "|        5634|       392|          4|    2383|2020-03-22 03:31:55|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":430516...|  null|\n",
       "|        5633|       392|          4|    2383|2020-03-22 03:31:32|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":430514...|  null|\n",
       "|        5632|       392|          4|    2383|2020-03-22 00:37:12|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":430507...|  null|\n",
       "|        5631|       392|          4|    2383|2020-03-22 00:25:34|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":430500...|  null|\n",
       "|        5630|       392|          4|    2383|2020-03-22 00:22:29|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":430493...|  null|\n",
       "|        5629|       392|          4|    2383|2020-03-22 00:20:24|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":430486...|  null|\n",
       "|        5628|       392|          4|    2383|2020-03-22 00:16:29|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":430479...|  null|\n",
       "|        5627|       392|          4|    2383|2020-03-22 00:14:58|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":430471...|  null|\n",
       "|        5626|       392|          4|    2383|2020-03-22 00:05:57|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":422275...|  null|\n",
       "|        5625|       392|          4|    2383|2020-03-22 00:05:22|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":420220...|  null|\n",
       "|        5624|       392|          4|    2383|2020-03-22 00:04:07|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":416118...|  null|\n",
       "|        5623|       392|          4|    2383|2020-03-21 23:54:21|             2|false|     M|      null|90d81642-6276-483...|            1|1961-10-01|[{\"obs_id\":387452...|  null|\n",
       "|        4336|       394|          1|    1836|2017-01-18 10:50:03|             5|false|     M|      null|c0cbe231-deb8-4cf...|            1|1972-12-07|[{\"obs_id\":35662,...|  null|\n",
       "|        4335|       394|          7|    1836|2017-01-18 10:01:03|             2|false|     M|      null|c0cbe231-deb8-4cf...|            1|1972-12-07|[{\"obs_id\":35654,...|  null|\n",
       "|        4950|       444|          1|    2094|2017-01-17 11:09:58|             5|false|     M|      null|a793f92a-87d3-4c4...|            1|1956-04-07|[{\"obs_id\":40700,...|  null|\n",
       "+------------+----------+-----------+--------+-------------------+--------------+-----+------+----------+--------------------+-------------+----------+--------------------+------+\n",
       "only showing top 20 rows"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.functions import col\n",
    "df= spark.read.format(\"delta\").load(\"flat_obs_orders.delta\")\n",
    "df.sort(col(\"encounter_datetime\").desc())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>version</th><th>timestamp</th><th>userId</th><th>userName</th><th>operation</th><th>operationParameters</th><th>job</th><th>notebook</th><th>clusterId</th><th>readVersion</th><th>isolationLevel</th><th>isBlindAppend</th></tr>\n",
       "<tr><td>107</td><td>2020-03-22 04:20:01</td><td>null</td><td>null</td><td>DELETE</td><td>[predicate -&gt; [&quot;(...</td><td>null</td><td>null</td><td>null</td><td>106</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>106</td><td>2020-03-22 04:19:54</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>105</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>105</td><td>2020-03-22 04:16:40</td><td>null</td><td>null</td><td>DELETE</td><td>[predicate -&gt; [&quot;(...</td><td>null</td><td>null</td><td>null</td><td>104</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>104</td><td>2020-03-22 04:16:33</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>103</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>103</td><td>2020-03-22 04:15:40</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>102</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>102</td><td>2020-03-22 04:15:15</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>101</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>101</td><td>2020-03-22 04:12:44</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>100</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>100</td><td>2020-03-22 03:57:12</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>99</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>99</td><td>2020-03-22 03:56:08</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>98</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>98</td><td>2020-03-22 03:50:20</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>97</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>97</td><td>2020-03-22 03:49:26</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>96</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>96</td><td>2020-03-22 03:49:09</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>95</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>95</td><td>2020-03-22 03:42:24</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>94</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>94</td><td>2020-03-22 03:33:41</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>93</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>93</td><td>2020-03-22 03:32:51</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>92</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>92</td><td>2020-03-22 03:32:32</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>91</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>91</td><td>2020-03-22 03:32:10</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>90</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>90</td><td>2020-03-22 03:25:27</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>89</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>89</td><td>2020-03-22 03:12:13</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>88</td><td>null</td><td>false</td></tr>\n",
       "<tr><td>88</td><td>2020-03-22 03:11:53</td><td>null</td><td>null</td><td>MERGE</td><td>[predicate -&gt; (ta...</td><td>null</td><td>null</td><td>null</td><td>87</td><td>null</td><td>false</td></tr>\n",
       "</table>\n",
       "only showing top 20 rows\n"
      ],
      "text/plain": [
       "+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+--------------+-------------+\n",
       "|version|          timestamp|userId|userName|operation| operationParameters| job|notebook|clusterId|readVersion|isolationLevel|isBlindAppend|\n",
       "+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+--------------+-------------+\n",
       "|    107|2020-03-22 04:20:01|  null|    null|   DELETE|[predicate -> [\"(...|null|    null|     null|        106|          null|        false|\n",
       "|    106|2020-03-22 04:19:54|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|        105|          null|        false|\n",
       "|    105|2020-03-22 04:16:40|  null|    null|   DELETE|[predicate -> [\"(...|null|    null|     null|        104|          null|        false|\n",
       "|    104|2020-03-22 04:16:33|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|        103|          null|        false|\n",
       "|    103|2020-03-22 04:15:40|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|        102|          null|        false|\n",
       "|    102|2020-03-22 04:15:15|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|        101|          null|        false|\n",
       "|    101|2020-03-22 04:12:44|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|        100|          null|        false|\n",
       "|    100|2020-03-22 03:57:12|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|         99|          null|        false|\n",
       "|     99|2020-03-22 03:56:08|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|         98|          null|        false|\n",
       "|     98|2020-03-22 03:50:20|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|         97|          null|        false|\n",
       "|     97|2020-03-22 03:49:26|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|         96|          null|        false|\n",
       "|     96|2020-03-22 03:49:09|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|         95|          null|        false|\n",
       "|     95|2020-03-22 03:42:24|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|         94|          null|        false|\n",
       "|     94|2020-03-22 03:33:41|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|         93|          null|        false|\n",
       "|     93|2020-03-22 03:32:51|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|         92|          null|        false|\n",
       "|     92|2020-03-22 03:32:32|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|         91|          null|        false|\n",
       "|     91|2020-03-22 03:32:10|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|         90|          null|        false|\n",
       "|     90|2020-03-22 03:25:27|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|         89|          null|        false|\n",
       "|     89|2020-03-22 03:12:13|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|         88|          null|        false|\n",
       "|     88|2020-03-22 03:11:53|  null|    null|    MERGE|[predicate -> (ta...|null|    null|     null|         87|          null|        false|\n",
       "+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+--------------+-------------+\n",
       "only showing top 20 rows"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from delta.tables import *\n",
    "# Access the Delta Lake table\n",
    "deltaTable = DeltaTable.forPath(spark, \"flat_obs_orders.delta\")\n",
    "deltaTable.history()#.select('timestamp')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ds1= spark.readStream.format(\"delta\")\\\n",
    "  .option(\"checkpointLocation\", \"checkpoint/streaming-agg\")\\\n",
    "  .option(\"ignoreChanges\", \"true\")\\\n",
    "  .option(\"ignoreDeletes\", \"true\")\\\n",
    "  .load(\"flat_obs_orders.delta\").writeStream\\\n",
    "  .format(\"console\")\\\n",
    "  .start()\n",
    "\n",
    "ds1.awaitTermination()\n"
   ]
  },
  {
   "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
}