tsk_20190313_experiment2-p1-patient-selection.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Analysis 2-1: Data selection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Intitializing Scala interpreter ..."
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Spark Web UI available at http://localhost:4041\n",
       "SparkContext available as 'sc' (version = 2.3.2, master = local[*], app id = local-1556274831546)\n",
       "SparkSession available as 'spark'\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "res0: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@602e2208\n"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "//%load src/scala/spark_imports.scala"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "import org.apache.spark.sql._\n",
       "import org.apache.spark.sql.functions._\n",
       "import org.apache.spark.sql.types._\n",
       "import spark.implicits._\n"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.sql._\n",
    "import org.apache.spark.sql.functions._\n",
    "import org.apache.spark.sql.types._\n",
    "import spark.implicits._"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "===Prescription dataset===\n",
      "registering patients\n",
      "registering prescriptions\n",
      "registering drugs\n",
      "ok!\n",
      "===Hospitalization dataset===\n",
      "registering all\n",
      "registering patients\n",
      "registering prescriptions\n",
      "registering drugs\n",
      "ok!\n",
      "===NPR Elders===\n",
      "ok!\n",
      "+--------+--------------------------+-----------+\n",
      "|database|tableName                 |isTemporary|\n",
      "+--------+--------------------------+-----------+\n",
      "|        |elders                    |true       |\n",
      "|        |elders_drugs              |true       |\n",
      "|        |elders_patients           |true       |\n",
      "|        |elders_prescriptions      |true       |\n",
      "|        |npr_elders                |true       |\n",
      "|        |prescription_drugs        |true       |\n",
      "|        |prescription_patients     |true       |\n",
      "|        |prescription_prescriptions|true       |\n",
      "+--------+--------------------------+-----------+\n",
      "\n",
      "done!\n"
     ]
    }
   ],
   "source": [
    "%run src/scala/register_tables.scala\n",
    "register_tables(spark, \"spark-warehouse/\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "elders: org.apache.spark.sql.DataFrame = [id: string, birthyear: int ... 21 more fields]\n"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val elders = spark.sql(\"select * from elders\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Split the data by gender and whether they have hospitalizations or not"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "hospitalized: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [id: string, birthyear: int ... 21 more fields]\n",
       "unhospitalized: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [id: string, birthyear: int ... 21 more fields]\n",
       "npr: org.apache.spark.sql.DataFrame = [id: string, num_hospitalizations: bigint ... 10 more fields]\n",
       "npr_ids: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [id2: string]\n"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val hospitalized = elders.where(\"diff_utleveringdato is not null\")\n",
    "val unhospitalized = elders\n",
    "    .where(\"diff_utleveringdato is null\")\n",
    "    .where(\"death_timestamp is null\")\n",
    "\n",
    "val npr = spark.sql(\"select * from npr_elders\")\n",
    "val npr_ids = npr.select(\"id\")\n",
    "    .withColumnRenamed(\"id\",\"id2\")\n",
    "    .distinct\n",
    "//npr_ids.printSchema\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Hospitalized patients: Join with NPR (Hospitalization register) So that we get the set of IDS: \n",
    "$Hosp.IDs \\cap NPR.IDs$\n",
    "\n",
    "For both genders"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Registered functions: unhospitalized_selector, hospitalized_selector,  npr_selector\n",
      "ok\n"
     ]
    }
   ],
   "source": [
    "%run src/scala/data_selection.scala"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Male"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "male_hospitalized: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [id: string, birthyear: int ... 21 more fields]\n",
       "male_unhospitalized: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [id: string, birthyear: int ... 21 more fields]\n",
       "male_hosp_ids: org.apache.spark.sql.DataFrame = [id2: string]\n",
       "eligible_male_hosp_prescriptions: org.apache.spark.sql.DataFrame = [id: string, birthyear: int ... 3 more fields]\n",
       "eligible_male_hosp_npr: org.apache.spark.sql.DataFrame = [diffdager_inn: int, diffdager_ut: int ... 1 more field]\n",
       "male_hosp_patients_count: Long = 80833\n",
       "male_after_filtering_count: Long = 80833\n"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val male_hospitalized     = hospitalized.where('gender === 1)\n",
    "val male_unhospitalized   = unhospitalized.where('gender === 1)\n",
    "\n",
    "val male_hosp_ids = male_hospitalized\n",
    "    .select(\"id\")\n",
    "    .distinct\n",
    "    .join(npr_ids, 'id==='id2).drop(\"id\")\n",
    "\n",
    "val eligible_male_hosp_prescriptions = hospitalized_selector(\n",
    "        male_hospitalized\n",
    "            .join(male_hosp_ids, 'id === 'id2)\n",
    "    )\n",
    "val eligible_male_hosp_npr           = npr_selector(\n",
    "        npr.join(male_hosp_ids, 'id === 'id2)\n",
    "    )\n",
    "\n",
    "val male_hosp_patients_count   = male_hospitalized.select('id).distinct.count\n",
    "val male_after_filtering_count   = male_hosp_ids.count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Female"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "female_hospitalized: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [id: string, birthyear: int ... 21 more fields]\n",
       "female_unhospitalized: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [id: string, birthyear: int ... 21 more fields]\n",
       "female_hosp_ids: org.apache.spark.sql.DataFrame = [id2: string]\n",
       "eligible_female_hosp_prescriptions: org.apache.spark.sql.DataFrame = [id: string, birthyear: int ... 3 more fields]\n",
       "eligible_female_hosp_npr: org.apache.spark.sql.DataFrame = [diffdager_inn: int, diffdager_ut: int ... 1 more field]\n",
       "female_hosp_patients_count: Long = 91595\n",
       "female_after_filtering_count: Long = 91595\n"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val female_hospitalized   = hospitalized.where('gender === 2)\n",
    "val female_unhospitalized = unhospitalized.where('gender === 2)\n",
    "\n",
    "val female_hosp_ids = female_hospitalized\n",
    "    .select(\"id\")\n",
    "    .distinct\n",
    "    .join(npr_ids, 'id==='id2).drop(\"id\")\n",
    "\n",
    "val eligible_female_hosp_prescriptions = hospitalized_selector(\n",
    "        female_hospitalized\n",
    "            .join(female_hosp_ids, 'id === 'id2)\n",
    "    )\n",
    "val eligible_female_hosp_npr           = npr_selector(\n",
    "        npr.join(female_hosp_ids, 'id === 'id2)\n",
    "    )\n",
    "\n",
    "val female_hosp_patients_count = female_hospitalized.select('id).distinct.count \n",
    "val female_after_filtering_count = female_hosp_ids.count\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Save hospitalization datasets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "savedir: String = spark-warehouse/experiment2-data/\n"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val savedir = \"spark-warehouse/experiment2-data/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ok\n"
     ]
    }
   ],
   "source": [
    "eligible_female_hosp_npr\n",
    "    .write.mode(SaveMode.Overwrite).parquet(savedir+\"female/hosp_npr\")\n",
    "eligible_female_hosp_prescriptions\n",
    "    .write.mode(SaveMode.Overwrite).parquet(savedir+\"female/hosp_pres\")\n",
    "\n",
    "eligible_male_hosp_npr\n",
    "    .write.mode(SaveMode.Overwrite).parquet(savedir+\"male/hosp_npr\")\n",
    "//This definitely wasn't the root of all evil in my analyses\n",
    "//eligible_female_hosp_prescriptions\n",
    "eligible_male_hosp_prescriptions\n",
    "    .write.mode(SaveMode.Overwrite).parquet(savedir+\"male/hosp_pres\")\n",
    "println(\"ok\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Save Unhospitalized sets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ok\n"
     ]
    }
   ],
   "source": [
    "unhospitalized_selector(male_unhospitalized)\n",
    "    .write.mode(SaveMode.Overwrite).parquet(savedir+\"male/unhosp_pres\")\n",
    "unhospitalized_selector(female_unhospitalized)\n",
    "    .write.mode(SaveMode.Overwrite).parquet(savedir+\"female/unhosp_pres\")\n",
    "println(\"ok\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Convert elders active drug dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ok\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "elders_active_drugs: org.apache.spark.sql.DataFrame = [id: string, drugcode: string ... 2 more fields]\n"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val elders_active_drugs = spark.read\n",
    "    .option(\"header\",\"true\")\n",
    "    .csv(\"datasets/elders_drug_duration\").drop(\"_c0\")\n",
    "    .select(\n",
    "        $\"pasientlopenr\".as(\"id\")\n",
    "        ,$\"atckode\".as(\"drugcode\")\n",
    "        ,$\"treatment_start\".cast(\"int\")\n",
    "        ,$\"treatment_end\".cast(\"int\")\n",
    "    )\n",
    "\n",
    "elders_active_drugs\n",
    "    .repartition($\"id\")\n",
    "    .write\n",
    "    .mode(SaveMode.Overwrite)    \n",
    "    .parquet(\"spark-warehouse/experiment2-data/elders_drug_duration\")\n",
    "println(\"ok\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "spylon-kernel",
   "language": "scala",
   "name": "spylon-kernel"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "file_extension": ".scala",
   "help_links": [
    {
     "text": "MetaKernel Magics",
     "url": "https://github.com/calysto/metakernel/blob/master/metakernel/magics/README.md"
    }
   ],
   "mimetype": "text/x-scala",
   "name": "scala",
   "pygments_lexer": "scala",
   "version": "0.4.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}