preprocessing_prescription.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Cell to initialize the Spark runtime and import functionality"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import org.apache.spark.sql.functions._\n",
    "import org.apache.spark.sql.types._\n",
    "import org.apache.spark.sql._\n",
    "import spark.implicits._\n",
    "\n",
    "val basedir = \"spark-warehouse/prescription/\"\n",
    "sc.getConf.getAll\n",
    "// set to true to execute code to display tables\n",
    "val execExtras=false"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load the dataset\n",
    "- Delete rows where patient has no ID ( PasientUtenID == 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "val path = \"datasets/prescription\"\n",
    "val raw= spark.read.option(\"delimiter\",\";\").option(\"header\",\"true\").csv(path)\n",
    "\n",
    "\n",
    "val converted = raw\n",
    "    .where(col(\"PasientUtenID\").notEqual(\"1\"))\n",
    "    .select(\n",
    "        $\" PasientLopeNr\".as(\"id\")\n",
    "        ,$\"PasientFodtAr\".cast(\"integer\").as(\"birthyear\")\n",
    "        ,$\"PasientKjonn\".cast(\"integer\").as(\"gender\")\n",
    "        ,unix_timestamp($\"UtleveringsDato\", \"yyyy.MM.dd\").cast(TimestampType).as(\"timestamp\")\n",
    "        ,$\"ATCKode\".as(\"drugcode\")\n",
    "    )\n",
    "    .where(\"id is not null\")\n",
    "    .where(\"birthyear is not null\")\n",
    "    .where(\"gender is not null\")\n",
    "    .where(\"timestamp is not null\")\n",
    "    .where(\"drugcode is not null\")\n",
    "    .cache\n",
    "  \n",
    " \n",
    "//val fields = raw.filter(col(\"PasientUtenID\").notEqual(\"1\"))\n",
    "\n",
    "//println(\"filtered away rows: \", raw.count-fields.count)\n",
    "if (execExtras) {\n",
    "converted.show(5)    \n",
    "}\n",
    "converted.printSchema"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create Prescription Dataframe, taking into consideration only prescriptions and the patient's ID\n",
    "* Take Patient ID, Date of prescription and Drug code(ATC Code)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "val prescriptions = converted.select(\n",
    "    $\"id\"\n",
    "    ,$\"timestamp\"\n",
    "    ,$\"drugcode\"\n",
    ")\n",
    "if (execExtras){\n",
    "    prescriptions.show(5)    \n",
    "}\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create Patients Dataframe, carrying info of people\n",
    "* We only have the ID, Birthyear and Gender"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "val patients = converted.select(\n",
    "    $\"id\"\n",
    "    ,$\"birthyear\"\n",
    "    ,$\"gender\"\n",
    ").distinct\n",
    "\n",
    "if (execExtras){\n",
    "    println(\"Total unique people in dataset\", patients.count)\n",
    "    patients.show(5)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Drugs in dataset\n",
    "* Contains 855 unique ATC codes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "val drugs = converted.select($\"drugcode\").distinct.orderBy(asc(\"drugcode\"))\n",
    "\n",
    "if (execExtras){\n",
    "    drugs.show(5)\n",
    "    drugs.count    \n",
    "}\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Save tables\n",
    "* Should be saved in the \"spark-warehouse/prescriptions/\" folder from the root directory"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "patients.write\n",
    "    .mode(SaveMode.Overwrite)\n",
    "    .parquet(basedir+\"patients\")\n",
    "println(\"ok\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "prescriptions.write\n",
    "    .mode(SaveMode.Overwrite)\n",
    "    .parquet(basedir+\"prescriptions\")\n",
    "println(\"ok\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "drugs.write\n",
    "    .mode(SaveMode.Overwrite)\n",
    "    .parquet(basedir+\"drugs\")\n",
    "println(\"ok\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if (execExtras){\n",
    "    patients.where(col(\"birthyear\")<1880).show\n",
    "}"
   ]
  }
 ],
 "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
}