hiv_summary_dataset.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# Install missing python packages\n",
    "import sys\n",
    "!{sys.executable} -m pip install -r requirements.txt\n",
    "\n",
    "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": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# import and optimize\n",
    "spark.sql(\"DROP TABLE IF EXISTS flat_obs\")\n",
    "spark.sql(\"CREATE TABLE flat_obs USING DELTA LOCATION 'flat_obs_orders.delta'\")   \n",
    "df = spark.sql(\"select * from flat_obs\")\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "%%time\n",
    "import pyspark.sql.functions as f\n",
    "nested_df = df.withColumn(\"obs\", f.explode_outer(\"obs\"))\\\n",
    "                .withColumn(\"orders\", f.explode_outer(\"orders\"))\n",
    "\n",
    "nested_df.createOrReplaceTempView(\"obs\")\n",
    "\n",
    "nested_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "started app\n",
      "+----------+------------+------------+-------------------+-----------+------+---------+----------+---------------------+--------------------+-----------+-------------------+----------------------+---------------------+------------------+\n",
      "|patient_id|patient_uuid|encounter_id| encounter_datetime|location_id|gender|birthdate|death_date|is_clinical_encounter|       analysis_date|out_of_care|    enrollment_date|enrollment_location_id|arv_start_location_id|viral_load_ordered|\n",
      "+----------+------------+------------+-------------------+-----------+------+---------+----------+---------------------+--------------------+-----------+-------------------+----------------------+---------------------+------------------+\n",
      "|     13460|        null|     6932376|2017-06-07 10:48:43|         13|     M|     null|      null|                    1|2018-06-27 13:22:...|       null|2017-06-07 10:48:43|                    13|                   13|                 1|\n",
      "|     13460|        null|     7131079|2017-09-14 08:45:08|         13|     M|     null|      null|                    1|2018-06-27 13:22:...|       null|2017-06-07 10:48:43|                    13|                   13|                 1|\n",
      "|     13460|        null|     7131081|2017-09-14 08:45:08|         13|     M|     null|      null|                    1|2018-06-27 13:22:...|       null|2017-06-07 10:48:43|                    13|                   13|                 1|\n",
      "|     13460|        null|     7131080|2017-09-14 08:45:08|         13|     M|     null|      null|                    1|2018-06-27 13:22:...|       null|2017-06-07 10:48:43|                    13|                   13|                 1|\n",
      "|     13460|        null|     7131082|2017-09-14 08:45:08|         13|     M|     null|      null|                    1|2018-06-27 13:22:...|       null|2017-06-07 10:48:43|                    13|                   13|                 1|\n",
      "|     13460|        null|     7131088|2017-09-14 08:45:08|         13|     M|     null|      null|                    1|2018-06-27 13:22:...|       null|2017-06-07 10:48:43|                    13|                   13|                 1|\n",
      "|     13460|        null|     7131084|2017-09-14 08:45:08|         13|     M|     null|      null|                    1|2018-06-27 13:22:...|       null|2017-06-07 10:48:43|                    13|                   13|                 1|\n",
      "|     13460|        null|     7131085|2017-09-14 08:45:08|         13|     M|     null|      null|                    1|2018-06-27 13:22:...|       null|2017-06-07 10:48:43|                    13|                   13|                 1|\n",
      "|     13460|        null|     7131087|2017-09-14 08:45:08|         13|     M|     null|      null|                    1|2018-06-27 13:22:...|       null|2017-06-07 10:48:43|                    13|                   13|                 1|\n",
      "|     13460|        null|     7131089|2017-09-14 08:45:08|         13|     M|     null|      null|                    1|2018-06-27 13:22:...|       null|2017-06-07 10:48:43|                    13|                   13|                 1|\n",
      "|     13460|        null|     7581252|2018-04-09 10:56:30|         13|     M|     null|      null|                    1|2018-06-27 13:22:...|       null|2017-06-07 10:48:43|                    13|                   13|                 1|\n",
      "+----------+------------+------------+-------------------+-----------+------+---------+----------+---------------------+--------------------+-----------+-------------------+----------------------+---------------------+------------------+\n",
      "\n",
      "Finished: Wed Jun 27 13:34:49 2018\n",
      "Took 746.355708 seconds\n"
     ]
    }
   ],
   "source": [
    "start_time = datetime.datetime.utcnow()\n",
    "print(\"started app\")\n",
    "\n",
    "\n",
    "\n",
    "# transformations\n",
    "spark.sql(\"\"\"select\n",
    "            *,\n",
    "            case \n",
    "                when flattened_obs.concept_id = 9082 and flattened_obs.value = '9036' then 'negative'\n",
    "                else null end as patient_care_status,\n",
    "            case \n",
    "                when flattened_obs.concept_id = 5303 and flattened_obs.value= '822' then 'exposed'\n",
    "                when flattened_obs.concept_id = 5303 and flattened_obs.value = '664' then 'negative'\n",
    "                when flattened_obs.concept_id = 5303 and flattened_obs.value = '1067' then 'unknown'\n",
    "                else null end as child_hiv_status,\n",
    "            case\n",
    "              when encounter_type in (1,2,3,4,10,14,15,17,19,26,32,\n",
    "              33,34,47,105,106,112,113,114,117,120,127,128,129,138,153,154,158) then 1\n",
    "            else null end as is_clinical_encounter\n",
    "            from obs\n",
    "        \"\"\").createOrReplaceTempView(\"hiv_summary_stage_0\")\n",
    "\n",
    "spark.sql(\"\"\"select \n",
    "             *,\n",
    "             case \n",
    "                 when flattened_obs.concept_id = 7013 and flattened_obs.value is not null then to_date(flattened_obs.value)\n",
    "                 when flattened_obs.concept_id = 7015 and flattened_obs.value is not null then to_date(flattened_obs.value)\n",
    "                 when encounter_type not in (21,99999) then encounter_datetime\n",
    "                 else null end as enrollment_date,\n",
    "             case \n",
    "                 when flattened_obs.concept_id = 1946 and flattened_obs.value = \"1065\" then 1\n",
    "                 when flattened_obs.concept_id = 1285 and flattened_obs.value in (\"1287\",\"9068\") then 1\n",
    "                 when flattened_obs.concept_id = 1596 then 1\n",
    "                 when flattened_obs.concept_id = 9082 and flattened_obs.value in (\"159\",\"9036\",\"9083\",\"1287\",\"9068\",\"9079\", \"9504\", \"1285\") then 1\n",
    "                 when encounter_type = 31 then 1\n",
    "                 else null end as out_of_care,\n",
    "             case \n",
    "                 when flattened_orders.order_concept_id = 856 then 1\n",
    "                 else null end as viral_load_ordered,\n",
    "             case \n",
    "                 when flattened_obs.concept_id = 1255 and flattened_obs.value is not null then location_id\n",
    "                 when flattened_obs.concept_id IN (1250, 1088, 2154) then location_id\n",
    "                 else null end as arv_location_id\n",
    "             from hiv_summary_stage_0\n",
    "             where \n",
    "                 child_hiv_status is null \n",
    "                 and patient_care_status is null \n",
    "                 and is_clinical_encounter = 1 \n",
    "             \"\"\").createOrReplaceTempView(\"hiv_summary_stage_1\")\n",
    "\n",
    "\n",
    "spark.sql(\"\"\"select patient_id,\n",
    "                    encounter_id,\n",
    "                    first(location_id) as location_id,\n",
    "                    first(encounter_datetime) as encounter_datetime,\n",
    "                    first(encounter_type) as encounter_type,\n",
    "                    first(enrollment_date, true) as enrollment_date,\n",
    "                    first(out_of_care, true) as out_of_care,\n",
    "                    first(location_id) as enrollment_location_id,\n",
    "                    first(viral_load_ordered, true) as viral_load_ordered,\n",
    "                    first(arv_location_id, true) as arv_location_id,\n",
    "                    first(gender) as gender,\n",
    "                    first(birthdate) as birthdate,\n",
    "                    first(death_date) as death_date,\n",
    "                    first(is_clinical_encounter) as is_clinical_encounter\n",
    "                    from hiv_summary_stage_1\n",
    "                    group by patient_id, encounter_id\n",
    "        \"\"\").createOrReplaceTempView(\"hiv_summary_stage_2\")\n",
    "\n",
    "hiv_summary_3 = spark.sql(\"\"\"select patient_id,\n",
    "                    null as patient_uuid,\n",
    "                    encounter_id,\n",
    "                    encounter_datetime,\n",
    "                    location_id,\n",
    "                    gender,\n",
    "                    birthdate,\n",
    "                    death_date,\n",
    "                    is_clinical_encounter,\n",
    "                    current_timestamp() as analysis_date,\n",
    "                    first(out_of_care, true) over p as out_of_care,\n",
    "                    first(enrollment_date, true) over p as enrollment_date,\n",
    "                    case when enrollment_date is not null then location_id\n",
    "                    else null end as enrollment_location_id,\n",
    "                    first(arv_location_id, true) over p as arv_start_location_id,\n",
    "                    viral_load_ordered\n",
    "                    from hiv_summary_stage_2\n",
    "                    window p as (partition by patient_id order by encounter_datetime)\n",
    "        \"\"\")\n",
    "\n",
    "hiv_summary_3.createOrReplaceTempView(\"hiv_summary_stage_3\")\n",
    "\n",
    "hiv_summary = spark.sql(\"\"\"select patient_id,\n",
    "                    null as patient_uuid,\n",
    "                    encounter_id,\n",
    "                    encounter_datetime,\n",
    "                    location_id,\n",
    "                    gender,\n",
    "                    birthdate,\n",
    "                    death_date,\n",
    "                    is_clinical_encounter,\n",
    "                    analysis_date,\n",
    "                    out_of_care,\n",
    "                    enrollment_date,\n",
    "                    first(enrollment_location_id, true) over p as enrollment_location_id,\n",
    "                    arv_start_location_id,\n",
    "                    viral_load_ordered\n",
    "                    from hiv_summary_stage_3\n",
    "                    window p as (partition by patient_id order by encounter_datetime)\n",
    "        \"\"\")\n",
    "\n",
    "hiv_summary.createOrReplaceTempView(\"hiv_sum\")\n",
    "\n",
    "\n",
    "#hiv_summary = spark.sql(\"select *,  from hiv_sum\")\n",
    "#spark.sql(\"select * from hiv_sum where ordered_viral_load is not null\").show(50)\n",
    "spark.sql(\"select * from hiv_sum where patient_id = 13460\").show(50)\n",
    "\n",
    "#print(hiv_summary.count())\n",
    "load_to_cassandra(hiv_summary)\n",
    "\n",
    "end_time = datetime.datetime.utcnow()\n",
    "print(\"Finished: \" + time.ctime()) \n",
    "print(\"Took {0} seconds\".format((end_time - start_time).total_seconds()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "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.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}