Notebook 2 - FHIR Spark LPR.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"https://www.ibm.com/watson/health/ai-stories/assets/images/ibm-watson-health-logo.png\" style=\"float: left; width: 40%; margin-bottom: 0.5em;\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Building Longitudinal Patient Record with Spark on FHIR\n",
    "\n",
    "Author: **Nick Kadochnikov** <nkadochn@us.ibm.com>\n",
    "\n",
    "[Section 1: Set-up cloud object storage function and credentials](#section_1)  \n",
    "[Section 2: Condition resource](#section_2)  \n",
    "[Section 3: Observation resource](#section_3)  \n",
    "[Section 4: Patient resource](#section_4)  \n",
    "[Section 5: Linking Conditions with Patients](#section_5)  \n",
    "[Section 6: Creating LPR](#section_6)  \n",
    "[Section 7: Saving LPR into COS](#section_7)  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Waiting for a Spark session to start...\n",
      "Spark Initialization Done! ApplicationId = app-20201114154030-0001\n",
      "KERNEL_ID = f666a35b-6594-48f3-8bb1-c04efe258091\n"
     ]
    }
   ],
   "source": [
    "import json\n",
    "import requests\n",
    "import os\n",
    "import types\n",
    "import pandas as pd\n",
    "import ibm_boto3\n",
    "import glob\n",
    "from ibm_botocore.client import Config\n",
    "from pprint import pprint\n",
    "import shutil\n",
    "\n",
    "import ibmos2spark\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import *\n",
    "\n",
    "pd.set_option('display.max_columns', None)\n",
    "pd.set_option('display.max_colwidth', 500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.conf.set(\"spark.sql.repl.eagerEval.enabled\",True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='section_1'></a>\n",
    "## Section 1. Set-up COS Functions and Credentials"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Define COS credentials"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Temporary credentials provided for DevDays only\n",
    "\n",
    "synthetic_mass_read_only = \\\n",
    "{\n",
    "  \"apikey\": \"HNJj8lVRmT-wX-n3ns2d8A8_iLFITob7ibC6aH66GZQX\",\n",
    "  \"endpoints\": \"https://control.cloud-object-storage.cloud.ibm.com/v2/endpoints\",\n",
    "  \"iam_apikey_description\": \"Auto-generated for key 418c8c60-5c31-4ed0-8a08-0f6641a01d46\",\n",
    "  \"iam_apikey_name\": \"dev_days\",\n",
    "  \"iam_role_crn\": \"crn:v1:bluemix:public:iam::::serviceRole:Reader\",\n",
    "  \"iam_serviceid_crn\": \"crn:v1:bluemix:public:iam-identity::a/f0dfe396162db060e2e2a53ff465dfa0::serviceid:ServiceId-e13864d8-8b73-4901-8060-b84123e5ca1c\",\n",
    "  \"resource_instance_id\": \"crn:v1:bluemix:public:cloud-object-storage:global:a/f0dfe396162db060e2e2a53ff465dfa0:3067bed7-8108-4d6e-ba32-5d5f643700e5::\"\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "cos_api_key = synthetic_mass_read_only\n",
    "input_bucket = 'syntheticmass-south'\n",
    "\n",
    "# https://s3.private.dal.us.cloud-object-storage.appdomain.cloud\n",
    "# https://s3.private.us-south.cloud-object-storage.appdomain.cloud\n",
    "\n",
    "credentials = {\n",
    "    'service_id': cos_api_key['iam_serviceid_crn'],\n",
    "    'api_key': cos_api_key['apikey'],\n",
    "    'endpoint': 'https://s3.private.us-south.cloud-object-storage.appdomain.cloud',\n",
    "    'iam_service_endpoint': 'https://iam.ng.bluemix.net/oidc/token'\n",
    "}\n",
    "\n",
    "configuration_name = 'syntheticmass-read' #Must be unique for each bucket / configuration!\n",
    "spark_cos = ibmos2spark.CloudObjectStorage(sc, credentials, configuration_name, 'bluemix_cos')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "boto_cos = ibm_boto3.resource('s3',\n",
    "                         ibm_api_key_id=cos_api_key['apikey'],\n",
    "                         ibm_service_instance_id=cos_api_key['iam_serviceid_crn'],\n",
    "                         ibm_auth_endpoint=credentials['iam_service_endpoint'],\n",
    "                         config=Config(signature_version='oauth'),\n",
    "                         endpoint_url=credentials['endpoint'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Set-up COS Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "def copy_to_local(cos, bucket_name, file_name, local_file_name):\n",
    "    try:\n",
    "        print(\"Downloading {0} from Bucket {1}\".format(file_name, bucket_name))\n",
    "        cos.Bucket(bucket_name).download_file(Key=file_name, Filename=local_file_name)\n",
    "        print(\"{0} downloaded\".format(local_file_name))\n",
    "    except Exception as e:\n",
    "        print(\"GENERAL ERROR: {0}\".format(e))\n",
    "        \n",
    "\n",
    "def copy_from_local(local_file_name, cos, bucket_name, file_name):\n",
    "    try:\n",
    "        print(\"Uploading {0}\".format(local_file_name))\n",
    "        cos.Bucket(bucket_name).upload_file(Key=file_name, Filename=local_file_name)\n",
    "        print(\"{0} uploaded to Bucket {1}\".format(file_name, bucket_name))\n",
    "    except Exception as e:\n",
    "        print(\"GENERAL ERROR: {0}\".format(e))\n",
    "        \n",
    "def get_bucket_contents(cos, bucket_name):\n",
    "    print(\"Retrieving bucket contents from: {0}\".format(bucket_name))\n",
    "    try:\n",
    "        files = cos.Bucket(bucket_name).objects.all()\n",
    "        for file in files:\n",
    "            print(\"Item: {0} ({1} bytes).\".format(file.key, file.size))\n",
    "    except ClientError as be:\n",
    "        print(\"CLIENT ERROR: {0}\\n\".format(be))\n",
    "    except Exception as e:\n",
    "        print(\"Unable to retrieve bucket contents: {0}\".format(e))\n",
    "        \n",
    "def create_text_file(cos, bucket_name, item_name, file_text):\n",
    "    print(\"Creating new item: {0}\".format(item_name))\n",
    "    try:\n",
    "        cos.Object(bucket_name, item_name).put(\n",
    "            Body=file_text\n",
    "        )\n",
    "        print(\"Item: {0} created!\".format(item_name))\n",
    "    except ClientError as be:\n",
    "        print(\"CLIENT ERROR: {0}\\n\".format(be))\n",
    "    except Exception as e:\n",
    "        print(\"Unable to create text file: {0}\".format(e))\n",
    "        \n",
    "def copy_from_local_with_time(df,resource, bucket):\n",
    "    today = date.today()\n",
    "    current_date = today.strftime(\"%d/%m/%Y\")\n",
    "\n",
    "    # create file name\n",
    "    file_name = \"{date}_{df_name}\".format(date = current_date,df_name = df)\n",
    "    \n",
    "    #push to COS\n",
    "    copy_from_local('./{}'.format(df), resource, bucket, file_name)\n",
    "    \n",
    "def delete_all_files(cos, bucket_name, should_contain=''):\n",
    "    files = cos.Bucket(bucket_name).objects.all()\n",
    "    for file in files:\n",
    "        if should_contain in file.key:\n",
    "            file.delete()\n",
    "            print(\"DELETED: {0}\".format(file.key))    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### COS List Files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "cos_fhir_read_bucket = input_bucket\n",
    "\n",
    "# get_bucket_contents(boto_cos, cos_fhir_read_bucket)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='section_2'></a>\n",
    "## Section 2. Explore \"Condition\" resource"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 12 ms, sys: 4 ms, total: 16 ms\n",
      "Wall time: 22 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "880043"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "input_file = 'r4_Condition.ndjson'\n",
    "\n",
    "\n",
    "%time condition = spark.read\\\n",
    "    .option('multiline','false')\\\n",
    "    .format('json')\\\n",
    "  .load(spark_cos.url(input_file, input_bucket))\n",
    "\n",
    "condition.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>abatementDateTime</th><th>clinicalStatus</th><th>code</th><th>encounter</th><th>id</th><th>onsetDateTime</th><th>recordedDate</th><th>resourceType</th><th>subject</th><th>verificationStatus</th></tr>\n",
       "<tr><td>null</td><td>[[[active, http:/...</td><td>[[[38341003, Hype...</td><td>[urn:uuid:a1de588...</td><td>97af11ac-672f-4ee...</td><td>1998-03-07T21:14:...</td><td>1998-03-07T21:14:...</td><td>Condition</td><td>[urn:uuid:9e0ed03...</td><td>[[[confirmed, htt...</td></tr>\n",
       "<tr><td>2009-10-29T21:14:...</td><td>[[[resolved, http...</td><td>[[[10509002, Acut...</td><td>[urn:uuid:c516e14...</td><td>f90d73a9-00fc-49a...</td><td>2009-10-22T21:14:...</td><td>2009-10-22T21:14:...</td><td>Condition</td><td>[urn:uuid:9e0ed03...</td><td>[[[confirmed, htt...</td></tr>\n",
       "<tr><td>2009-11-04T21:14:...</td><td>[[[resolved, http...</td><td>[[[444814009, Vir...</td><td>[urn:uuid:54d8f6e...</td><td>f21db262-d11f-42f...</td><td>2009-10-28T21:14:...</td><td>2009-10-28T21:14:...</td><td>Condition</td><td>[urn:uuid:9e0ed03...</td><td>[[[confirmed, htt...</td></tr>\n",
       "<tr><td>2010-06-28T21:14:...</td><td>[[[resolved, http...</td><td>[[[62106007, Conc...</td><td>[urn:uuid:fddf27f...</td><td>de112595-e70d-416...</td><td>2010-04-29T21:14:...</td><td>2010-04-29T21:14:...</td><td>Condition</td><td>[urn:uuid:9e0ed03...</td><td>[[[confirmed, htt...</td></tr>\n",
       "<tr><td>2017-07-20T07:18:...</td><td>[[[resolved, http...</td><td>[[[444814009, Vir...</td><td>[urn:uuid:e3dd3e3...</td><td>b677ae13-cba5-4c5...</td><td>2017-07-06T07:18:...</td><td>2017-07-06T07:18:...</td><td>Condition</td><td>[urn:uuid:6fe098b...</td><td>[[[confirmed, htt...</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+------------+--------------------+--------------------+\n",
       "|   abatementDateTime|      clinicalStatus|                code|           encounter|                  id|       onsetDateTime|        recordedDate|resourceType|             subject|  verificationStatus|\n",
       "+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+------------+--------------------+--------------------+\n",
       "|                null|[[[active, http:/...|[[[162864005, Bod...|[urn:uuid:23540f0...|687bc0a3-8e0f-467...|1971-11-25T19:33:...|1971-11-25T19:33:...|   Condition|[urn:uuid:dfe7387...|[[[confirmed, htt...|\n",
       "|2013-06-19T01:28:...|[[[resolved, http...|[[[72892002, Norm...|[urn:uuid:7c1d010...|3e679abf-5754-4e6...|2012-11-21T01:28:...|2012-11-21T01:28:...|   Condition|[urn:uuid:c75ff1a...|[[[confirmed, htt...|\n",
       "|                null|[[[active, http:/...|[[[408512008, Bod...|[urn:uuid:5c5f5ec...|14de3fe3-846b-4b8...|1987-09-03T19:33:...|1987-09-03T19:33:...|   Condition|[urn:uuid:dfe7387...|[[[confirmed, htt...|\n",
       "|                null|[[[active, http:/...|[[[239873007, Ost...|[urn:uuid:d02ec16...|375f6611-34df-48f...|1987-11-16T19:33:...|1987-11-16T19:33:...|   Condition|[urn:uuid:dfe7387...|[[[confirmed, htt...|\n",
       "|                null|[[[active, http:/...|[[[55822004, Hype...|[urn:uuid:d58a741...|9742f49d-0ec7-4ce...|2000-12-07T19:33:...|2000-12-07T19:33:...|   Condition|[urn:uuid:dfe7387...|[[[confirmed, htt...|\n",
       "+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+------------+--------------------+--------------------+"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# condition = condition.limit(100000).cache()\n",
    "condition.limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- abatementDateTime: string (nullable = true)\n",
      " |-- clinicalStatus: struct (nullable = true)\n",
      " |    |-- coding: array (nullable = true)\n",
      " |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |-- code: string (nullable = true)\n",
      " |    |    |    |-- system: string (nullable = true)\n",
      " |-- code: struct (nullable = true)\n",
      " |    |-- coding: array (nullable = true)\n",
      " |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |-- code: string (nullable = true)\n",
      " |    |    |    |-- display: string (nullable = true)\n",
      " |    |    |    |-- system: string (nullable = true)\n",
      " |    |-- text: string (nullable = true)\n",
      " |-- encounter: struct (nullable = true)\n",
      " |    |-- reference: string (nullable = true)\n",
      " |-- id: string (nullable = true)\n",
      " |-- onsetDateTime: string (nullable = true)\n",
      " |-- recordedDate: string (nullable = true)\n",
      " |-- resourceType: string (nullable = true)\n",
      " |-- subject: struct (nullable = true)\n",
      " |    |-- reference: string (nullable = true)\n",
      " |-- verificationStatus: struct (nullable = true)\n",
      " |    |-- coding: array (nullable = true)\n",
      " |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |-- code: string (nullable = true)\n",
      " |    |    |    |-- system: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "condition.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Identifying all patients with Diabetes in Confirmed status in the last X years"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Record selection {Option-1} preserving names"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>code</th><th>code</th><th>code</th><th>display</th><th>first_observation_date</th></tr>\n",
       "<tr><td>22bb9254-8abb-467...</td><td>[active]</td><td>[confirmed]</td><td>[44054006]</td><td>[Diabetes]</td><td>1980-06-19</td></tr>\n",
       "<tr><td>7c485227-4e00-42d...</td><td>[active]</td><td>[confirmed]</td><td>[44054006]</td><td>[Diabetes]</td><td>1993-06-24</td></tr>\n",
       "<tr><td>4c65f823-e939-417...</td><td>[active]</td><td>[confirmed]</td><td>[44054006]</td><td>[Diabetes]</td><td>2013-06-08</td></tr>\n",
       "<tr><td>a851e37f-f93e-421...</td><td>[active]</td><td>[confirmed]</td><td>[44054006]</td><td>[Diabetes]</td><td>1985-07-20</td></tr>\n",
       "<tr><td>40c9e21b-9a15-475...</td><td>[active]</td><td>[confirmed]</td><td>[44054006]</td><td>[Diabetes]</td><td>2011-09-01</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+--------+-----------+----------+----------+----------------------+\n",
       "|          patient_id|    code|       code|      code|   display|first_observation_date|\n",
       "+--------------------+--------+-----------+----------+----------+----------------------+\n",
       "|9683a6fb-c277-418...|[active]|[confirmed]|[44054006]|[Diabetes]|            2017-07-06|\n",
       "|f4cc08a7-15e3-47a...|[active]|[confirmed]|[44054006]|[Diabetes]|            1957-07-09|\n",
       "|d9c0c67c-ad91-4c1...|[active]|[confirmed]|[44054006]|[Diabetes]|            2019-03-30|\n",
       "|8ea4b567-9912-46c...|[active]|[confirmed]|[44054006]|[Diabetes]|            2011-06-16|\n",
       "|d3186727-4f34-4a6...|[active]|[confirmed]|[44054006]|[Diabetes]|            1966-03-01|\n",
       "+--------------------+--------+-----------+----------+----------+----------------------+"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "condition_diabetes = condition.\\\n",
    "where(array_contains('clinicalStatus.coding.code', 'active')).\\\n",
    "where(array_contains('verificationStatus.coding.code', 'confirmed')).\\\n",
    "where(array_contains('code.coding.code', '44054006')).\\\n",
    "where(condition.onsetDateTime > '1900-01-01').\\\n",
    "select(\\\n",
    "       (condition['subject.reference'].substr(10, 40).alias('patient_id')),\n",
    "       'clinicalStatus.coding.code',\\\n",
    "       'verificationStatus.coding.code',\\\n",
    "       'code.coding.code', \\\n",
    "       'code.coding.display',\\\n",
    "       to_date(condition['onsetDateTime']).alias('first_observation_date'))\n",
    "\n",
    "condition_diabetes.limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- patient_id: string (nullable = true)\n",
      " |-- code: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- code: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- code: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- display: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- first_observation_date: date (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "condition_diabetes.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Record selection {Option-2} with renaming"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>condition_status</th><th>verification_status</th><th>snomed_code</th><th>snomed_name</th><th>first_observation_date</th></tr>\n",
       "<tr><td>9683a6fb-c277-418...</td><td>[active]</td><td>[confirmed]</td><td>[44054006]</td><td>[Diabetes]</td><td>2017-07-06</td></tr>\n",
       "<tr><td>f4cc08a7-15e3-47a...</td><td>[active]</td><td>[confirmed]</td><td>[44054006]</td><td>[Diabetes]</td><td>1957-07-09</td></tr>\n",
       "<tr><td>d9c0c67c-ad91-4c1...</td><td>[active]</td><td>[confirmed]</td><td>[44054006]</td><td>[Diabetes]</td><td>2019-03-30</td></tr>\n",
       "<tr><td>8ea4b567-9912-46c...</td><td>[active]</td><td>[confirmed]</td><td>[44054006]</td><td>[Diabetes]</td><td>2011-06-16</td></tr>\n",
       "<tr><td>d3186727-4f34-4a6...</td><td>[active]</td><td>[confirmed]</td><td>[44054006]</td><td>[Diabetes]</td><td>1966-03-01</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----------------+-------------------+-----------+-----------+----------------------+\n",
       "|          patient_id|condition_status|verification_status|snomed_code|snomed_name|first_observation_date|\n",
       "+--------------------+----------------+-------------------+-----------+-----------+----------------------+\n",
       "|64601eeb-1631-455...|        [active]|        [confirmed]| [44054006]| [Diabetes]|            1976-11-05|\n",
       "|5601537d-266f-4ad...|        [active]|        [confirmed]| [44054006]| [Diabetes]|            1988-12-07|\n",
       "|6d21f847-9768-432...|        [active]|        [confirmed]| [44054006]| [Diabetes]|            1990-11-10|\n",
       "|6e3c0138-330f-496...|        [active]|        [confirmed]| [44054006]| [Diabetes]|            2011-06-26|\n",
       "|97d5a8e7-8f31-46b...|        [active]|        [confirmed]| [44054006]| [Diabetes]|            1998-03-20|\n",
       "+--------------------+----------------+-------------------+-----------+-----------+----------------------+"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "condition_diabetes = condition.\\\n",
    "where(array_contains('clinicalStatus.coding.code', 'active')).\\\n",
    "where(array_contains('verificationStatus.coding.code', 'confirmed')).\\\n",
    "where(array_contains('code.coding.code', '44054006')).\\\n",
    "where(condition.onsetDateTime > '1900-01-01').\\\n",
    "select(\\\n",
    "       (condition['subject.reference'].substr(10, 40).alias('patient_id')),\n",
    "       col('clinicalStatus.coding.code').alias('condition_status'),\\\n",
    "       col('verificationStatus.coding.code').alias('verification_status'),\\\n",
    "       col('code.coding.code').alias('snomed_code'), \\\n",
    "       col('code.coding.display').alias('snomed_name'),\\\n",
    "       to_date(condition['onsetDateTime']).alias('first_observation_date'))\n",
    "\n",
    "condition_diabetes.limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- patient_id: string (nullable = true)\n",
      " |-- condition_status: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- verification_status: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- snomed_code: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- snomed_name: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- first_observation_date: date (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "condition_diabetes.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Record selection {Option-3} with renaming and flattening"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>condition_status</th><th>verification_status</th><th>snomed_code</th><th>snomed_name</th><th>first_observation_date</th></tr>\n",
       "<tr><td>4557c356-5cd3-483...</td><td>active</td><td>confirmed</td><td>44054006</td><td>Diabetes</td><td>2018-10-25</td></tr>\n",
       "<tr><td>84ec9e91-2bc9-4be...</td><td>active</td><td>confirmed</td><td>44054006</td><td>Diabetes</td><td>1986-11-04</td></tr>\n",
       "<tr><td>6671cdcf-ea78-46f...</td><td>active</td><td>confirmed</td><td>44054006</td><td>Diabetes</td><td>2015-05-17</td></tr>\n",
       "<tr><td>ed4edac7-daf7-4db...</td><td>active</td><td>confirmed</td><td>44054006</td><td>Diabetes</td><td>2010-10-01</td></tr>\n",
       "<tr><td>68eaf7e6-05c9-4a2...</td><td>active</td><td>confirmed</td><td>44054006</td><td>Diabetes</td><td>2018-11-14</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----------------+-------------------+-----------+-----------+----------------------+\n",
       "|          patient_id|condition_status|verification_status|snomed_code|snomed_name|first_observation_date|\n",
       "+--------------------+----------------+-------------------+-----------+-----------+----------------------+\n",
       "|64601eeb-1631-455...|          active|          confirmed|   44054006|   Diabetes|            1976-11-05|\n",
       "|5601537d-266f-4ad...|          active|          confirmed|   44054006|   Diabetes|            1988-12-07|\n",
       "|6d21f847-9768-432...|          active|          confirmed|   44054006|   Diabetes|            1990-11-10|\n",
       "|6e3c0138-330f-496...|          active|          confirmed|   44054006|   Diabetes|            2011-06-26|\n",
       "|97d5a8e7-8f31-46b...|          active|          confirmed|   44054006|   Diabetes|            1998-03-20|\n",
       "+--------------------+----------------+-------------------+-----------+-----------+----------------------+"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "condition_diabetes = condition.\\\n",
    "where(array_contains('clinicalStatus.coding.code', 'active')).\\\n",
    "where(array_contains('verificationStatus.coding.code', 'confirmed')).\\\n",
    "where(array_contains('code.coding.code', '44054006')).\\\n",
    "where(condition.onsetDateTime > '1900-01-01').\\\n",
    "withColumn('condition_status', condition['clinicalStatus.coding.code'].getItem(0)).\\\n",
    "withColumn('verification_status', condition['verificationStatus.coding.code'].getItem(0)).\\\n",
    "withColumn('snomed_code', condition['code.coding.code'].getItem(0)).\\\n",
    "withColumn('snomed_name', condition['code.coding.display'].getItem(0)).\\\n",
    "select(\\\n",
    "       (condition['subject.reference'].substr(10, 40).alias('patient_id')),\n",
    "       'condition_status',\\\n",
    "       'verification_status',\\\n",
    "       'snomed_code', \\\n",
    "       'snomed_name',\\\n",
    "       to_date(condition['onsetDateTime']).alias('first_observation_date'))\n",
    "\n",
    "condition_diabetes.limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- patient_id: string (nullable = true)\n",
      " |-- condition_status: string (nullable = true)\n",
      " |-- verification_status: string (nullable = true)\n",
      " |-- snomed_code: string (nullable = true)\n",
      " |-- snomed_name: string (nullable = true)\n",
      " |-- first_observation_date: date (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "condition_diabetes.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='section_3'></a>\n",
    "## Section 3. Explore \"Observation\" resource"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 92 ms, sys: 20 ms, total: 112 ms\n",
      "Wall time: 1min 10s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "20515269"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "input_file = 'r4_Observation.ndjson'\n",
    "\n",
    "\n",
    "%time observation = spark.read\\\n",
    "    .option('multiline','false')\\\n",
    "    .format('json')\\\n",
    "  .load(spark_cos.url(input_file, input_bucket))\n",
    "\n",
    "observation.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>category</th><th>code</th><th>component</th><th>effectiveDateTime</th><th>encounter</th><th>id</th><th>issued</th><th>resourceType</th><th>status</th><th>subject</th><th>valueCodeableConcept</th><th>valueQuantity</th><th>valueString</th></tr>\n",
       "<tr><td>[[[[laboratory, l...</td><td>[[[785-6, MCH [En...</td><td>null</td><td>2018-11-20T13:31:...</td><td>[urn:uuid:ca8c667...</td><td>17fba280-0835-46e...</td><td>2018-11-20T13:31:...</td><td>Observation</td><td>final</td><td>[urn:uuid:f3650b2...</td><td>null</td><td>[pg, http://units...</td><td>null</td></tr>\n",
       "<tr><td>[[[[laboratory, l...</td><td>[[[786-4, MCHC [M...</td><td>null</td><td>2018-11-20T13:31:...</td><td>[urn:uuid:ca8c667...</td><td>61ef0100-4f06-4c0...</td><td>2018-11-20T13:31:...</td><td>Observation</td><td>final</td><td>[urn:uuid:f3650b2...</td><td>null</td><td>[g/dL, http://uni...</td><td>null</td></tr>\n",
       "<tr><td>[[[[laboratory, l...</td><td>[[[21000-5, Eryth...</td><td>null</td><td>2018-11-20T13:31:...</td><td>[urn:uuid:ca8c667...</td><td>35e723ef-afa3-4a7...</td><td>2018-11-20T13:31:...</td><td>Observation</td><td>final</td><td>[urn:uuid:f3650b2...</td><td>null</td><td>[fL, http://units...</td><td>null</td></tr>\n",
       "<tr><td>[[[[laboratory, l...</td><td>[[[777-3, Platele...</td><td>null</td><td>2018-11-20T13:31:...</td><td>[urn:uuid:ca8c667...</td><td>4b87b127-352e-419...</td><td>2018-11-20T13:31:...</td><td>Observation</td><td>final</td><td>[urn:uuid:f3650b2...</td><td>null</td><td>[10*3/uL, http://...</td><td>null</td></tr>\n",
       "<tr><td>[[[[laboratory, l...</td><td>[[[32207-3, Plate...</td><td>null</td><td>2018-11-20T13:31:...</td><td>[urn:uuid:ca8c667...</td><td>e90ae5c1-54fe-417...</td><td>2018-11-20T13:31:...</td><td>Observation</td><td>final</td><td>[urn:uuid:f3650b2...</td><td>null</td><td>[fL, http://units...</td><td>null</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+--------------------+---------+--------------------+--------------------+--------------------+--------------------+------------+------+--------------------+--------------------+--------------------+-----------+\n",
       "|            category|                code|component|   effectiveDateTime|           encounter|                  id|              issued|resourceType|status|             subject|valueCodeableConcept|       valueQuantity|valueString|\n",
       "+--------------------+--------------------+---------+--------------------+--------------------+--------------------+--------------------+------------+------+--------------------+--------------------+--------------------+-----------+\n",
       "|[[[[laboratory, l...|[[[2339-0, Glucos...|     null|2018-02-09T08:11:...|[urn:uuid:0d8d3bc...|3926f61b-8170-46f...|2018-02-09T08:11:...| Observation| final|[urn:uuid:a36eada...|                null|[mg/dL, http://un...|       null|\n",
       "|[[[[laboratory, l...|[[[6299-2, Urea N...|     null|2018-02-09T08:11:...|[urn:uuid:0d8d3bc...|380883fa-90a3-4c6...|2018-02-09T08:11:...| Observation| final|[urn:uuid:a36eada...|                null|[mg/dL, http://un...|       null|\n",
       "|[[[[laboratory, l...|[[[38483-4, Creat...|     null|2018-02-09T08:11:...|[urn:uuid:0d8d3bc...|13164a14-a945-4ad...|2018-02-09T08:11:...| Observation| final|[urn:uuid:a36eada...|                null|[mg/dL, http://un...|       null|\n",
       "|[[[[laboratory, l...|[[[49765-1, Calci...|     null|2018-02-09T08:11:...|[urn:uuid:0d8d3bc...|f30c7116-4a1e-415...|2018-02-09T08:11:...| Observation| final|[urn:uuid:a36eada...|                null|[mg/dL, http://un...|       null|\n",
       "|[[[[laboratory, l...|[[[2947-0, Sodium...|     null|2018-02-09T08:11:...|[urn:uuid:0d8d3bc...|f5f462a8-96ad-483...|2018-02-09T08:11:...| Observation| final|[urn:uuid:a36eada...|                null|[mmol/L, http://u...|       null|\n",
       "+--------------------+--------------------+---------+--------------------+--------------------+--------------------+--------------------+------------+------+--------------------+--------------------+--------------------+-----------+"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# observation = observation.limit(100000).cache()\n",
    "observation.limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- category: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- coding: array (nullable = true)\n",
      " |    |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |    |-- code: string (nullable = true)\n",
      " |    |    |    |    |-- display: string (nullable = true)\n",
      " |    |    |    |    |-- system: string (nullable = true)\n",
      " |-- code: struct (nullable = true)\n",
      " |    |-- coding: array (nullable = true)\n",
      " |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |-- code: string (nullable = true)\n",
      " |    |    |    |-- display: string (nullable = true)\n",
      " |    |    |    |-- system: string (nullable = true)\n",
      " |    |-- text: string (nullable = true)\n",
      " |-- component: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- code: struct (nullable = true)\n",
      " |    |    |    |-- coding: array (nullable = true)\n",
      " |    |    |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |    |    |-- code: string (nullable = true)\n",
      " |    |    |    |    |    |-- display: string (nullable = true)\n",
      " |    |    |    |    |    |-- system: string (nullable = true)\n",
      " |    |    |    |-- text: string (nullable = true)\n",
      " |    |    |-- valueQuantity: struct (nullable = true)\n",
      " |    |    |    |-- code: string (nullable = true)\n",
      " |    |    |    |-- system: string (nullable = true)\n",
      " |    |    |    |-- unit: string (nullable = true)\n",
      " |    |    |    |-- value: double (nullable = true)\n",
      " |-- effectiveDateTime: string (nullable = true)\n",
      " |-- encounter: struct (nullable = true)\n",
      " |    |-- reference: string (nullable = true)\n",
      " |-- id: string (nullable = true)\n",
      " |-- issued: string (nullable = true)\n",
      " |-- resourceType: string (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- subject: struct (nullable = true)\n",
      " |    |-- reference: string (nullable = true)\n",
      " |-- valueCodeableConcept: struct (nullable = true)\n",
      " |    |-- coding: array (nullable = true)\n",
      " |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |-- code: string (nullable = true)\n",
      " |    |    |    |-- display: string (nullable = true)\n",
      " |    |    |    |-- system: string (nullable = true)\n",
      " |    |-- text: string (nullable = true)\n",
      " |-- valueQuantity: struct (nullable = true)\n",
      " |    |-- code: string (nullable = true)\n",
      " |    |-- system: string (nullable = true)\n",
      " |    |-- unit: string (nullable = true)\n",
      " |    |-- value: double (nullable = true)\n",
      " |-- valueString: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "observation.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Record selection with renaming and flattening"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>loinc_code</th><th>loinc_name</th><th>effective_date</th></tr>\n",
       "<tr><td>d3d39284-0811-4c6...</td><td>4548-4</td><td>Hemoglobin A1c/He...</td><td>1995-03-25</td></tr>\n",
       "<tr><td>d3d39284-0811-4c6...</td><td>4548-4</td><td>Hemoglobin A1c/He...</td><td>1997-03-29</td></tr>\n",
       "<tr><td>d3d39284-0811-4c6...</td><td>4548-4</td><td>Hemoglobin A1c/He...</td><td>1999-04-03</td></tr>\n",
       "<tr><td>d3d39284-0811-4c6...</td><td>4548-4</td><td>Hemoglobin A1c/He...</td><td>2001-03-10</td></tr>\n",
       "<tr><td>d3d39284-0811-4c6...</td><td>4548-4</td><td>Hemoglobin A1c/He...</td><td>2002-03-16</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----------+--------------------+--------------+\n",
       "|          patient_id|loinc_code|          loinc_name|effective_date|\n",
       "+--------------------+----------+--------------------+--------------+\n",
       "|c5234d2f-ae48-47a...|    4548-4|Hemoglobin A1c/He...|    2010-02-22|\n",
       "|c5234d2f-ae48-47a...|    4548-4|Hemoglobin A1c/He...|    2011-02-28|\n",
       "|c5234d2f-ae48-47a...|    4548-4|Hemoglobin A1c/He...|    2012-03-05|\n",
       "|c5234d2f-ae48-47a...|    4548-4|Hemoglobin A1c/He...|    2013-03-11|\n",
       "|c5234d2f-ae48-47a...|    4548-4|Hemoglobin A1c/He...|    2014-03-17|\n",
       "+--------------------+----------+--------------------+--------------+"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "observation_diabetes = observation.\\\n",
    "where(array_contains('code.coding.code', '4548-4')).\\\n",
    "withColumn('loinc_code', observation['code.coding.code'].getItem(0)).\\\n",
    "withColumn('loinc_name', observation['code.coding.display'].getItem(0)).\\\n",
    "select(\\\n",
    "        (observation['subject.reference'].substr(10, 40).alias('patient_id')),\n",
    "        'loinc_code',\\\n",
    "        'loinc_name',\\\n",
    "        to_date(observation['effectiveDateTime']).alias('effective_date'))\n",
    "\n",
    "observation_diabetes.limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "####  Check the frequency of HbA1c tests"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 4 ms, sys: 4 ms, total: 8 ms\n",
      "Wall time: 34.6 ms\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>loinc_code</th><th>loinc_name</th><th>records_count</th></tr>\n",
       "<tr><td>506694f3-f741-466...</td><td>4548-4</td><td>Hemoglobin A1c/He...</td><td>10</td></tr>\n",
       "<tr><td>6ad12ea3-279d-436...</td><td>4548-4</td><td>Hemoglobin A1c/He...</td><td>7</td></tr>\n",
       "<tr><td>2cc1789c-4617-4f0...</td><td>4548-4</td><td>Hemoglobin A1c/He...</td><td>10</td></tr>\n",
       "<tr><td>bd2f2e15-8fb6-4f5...</td><td>4548-4</td><td>Hemoglobin A1c/He...</td><td>8</td></tr>\n",
       "<tr><td>62c4a7fa-105a-480...</td><td>4548-4</td><td>Hemoglobin A1c/He...</td><td>10</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----------+--------------------+-------------+\n",
       "|          patient_id|loinc_code|          loinc_name|records_count|\n",
       "+--------------------+----------+--------------------+-------------+\n",
       "|9a381f63-cc00-434...|    4548-4|Hemoglobin A1c/He...|           10|\n",
       "|3afa51e6-6c5a-4ac...|    4548-4|Hemoglobin A1c/He...|           11|\n",
       "|d6e3679a-cf2c-40d...|    4548-4|Hemoglobin A1c/He...|            4|\n",
       "|cdccbad9-fdde-45b...|    4548-4|Hemoglobin A1c/He...|           11|\n",
       "|6f8df8d4-7fca-4c3...|    4548-4|Hemoglobin A1c/He...|            8|\n",
       "+--------------------+----------+--------------------+-------------+"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%time loinc_code_freqency = observation_diabetes .groupby('patient_id', 'loinc_code', 'loinc_name').\\\n",
    "    agg(\\\n",
    "        count('effective_date').alias('records_count')\\\n",
    "       )\n",
    "\n",
    "loinc_code_freqency.where(loinc_code_freqency['records_count'] > 1).limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>summary</th><th>records_count</th></tr>\n",
       "<tr><td>count</td><td>38637</td></tr>\n",
       "<tr><td>mean</td><td>11.279861272873152</td></tr>\n",
       "<tr><td>stddev</td><td>14.05664672738714</td></tr>\n",
       "<tr><td>min</td><td>1</td></tr>\n",
       "<tr><td>max</td><td>113</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+-------+------------------+\n",
       "|summary|     records_count|\n",
       "+-------+------------------+\n",
       "|  count|             38637|\n",
       "|   mean|11.279861272873152|\n",
       "| stddev| 14.05664672738714|\n",
       "|    min|                 1|\n",
       "|    max|               113|\n",
       "+-------+------------------+"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "loinc_code_freqency.describe('records_count')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='section_4'></a>\n",
    "## Section 4.  Explore \"Patient\" resource"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 8 ms, sys: 0 ns, total: 8 ms\n",
      "Wall time: 3.02 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "117311"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "input_file = 'r4_Patient.ndjson'\n",
    "\n",
    "\n",
    "%time patient = spark.read\\\n",
    "    .option('multiline','false')\\\n",
    "    .format('json')\\\n",
    "  .load(spark_cos.url(input_file, input_bucket))\n",
    "\n",
    "patient.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>address</th><th>birthDate</th><th>communication</th><th>deceasedDateTime</th><th>extension</th><th>gender</th><th>id</th><th>identifier</th><th>maritalStatus</th><th>multipleBirthBoolean</th><th>multipleBirthInteger</th><th>name</th><th>resourceType</th><th>telecom</th><th>text</th></tr>\n",
       "<tr><td>[[Lowell, US, [[[...</td><td>1940-11-26</td><td>[[[[[en-US, Engli...</td><td>null</td><td>[[[[ombCategory, ...</td><td>female</td><td>6c5332a6-e49d-4df...</td><td>[[https://github....</td><td>[[[M, M, http://t...</td><td>false</td><td>null</td><td>[[Osinski784, [Ho...</td><td>Patient</td><td>[[phone, home, 55...</td><td>[&lt;div xmlns=&quot;http...</td></tr>\n",
       "<tr><td>[[Danvers, US, [[...</td><td>2013-09-06</td><td>[[[[[en-US, Engli...</td><td>null</td><td>[[[[ombCategory, ...</td><td>male</td><td>89d90dc7-1776-444...</td><td>[[https://github....</td><td>[[[S, Never Marri...</td><td>false</td><td>null</td><td>[[Fay398, [Hollis...</td><td>Patient</td><td>[[phone, home, 55...</td><td>[&lt;div xmlns=&quot;http...</td></tr>\n",
       "<tr><td>[[Westford, US, [...</td><td>1996-07-01</td><td>[[[[[en-US, Engli...</td><td>null</td><td>[[[[ombCategory, ...</td><td>female</td><td>aacee13c-aea8-45c...</td><td>[[https://github....</td><td>[[[S, Never Marri...</td><td>false</td><td>null</td><td>[[Klein929, [Anto...</td><td>Patient</td><td>[[phone, home, 55...</td><td>[&lt;div xmlns=&quot;http...</td></tr>\n",
       "<tr><td>[[Worcester, US, ...</td><td>1979-02-11</td><td>[[[[[es, Spanish,...</td><td>null</td><td>[[[[ombCategory, ...</td><td>male</td><td>e093f51c-562b-4d1...</td><td>[[https://github....</td><td>[[[M, M, http://t...</td><td>false</td><td>null</td><td>[[Aguayo104, [Dav...</td><td>Patient</td><td>[[phone, home, 55...</td><td>[&lt;div xmlns=&quot;http...</td></tr>\n",
       "<tr><td>[[Mansfield, US, ...</td><td>1984-11-28</td><td>[[[[[en-US, Engli...</td><td>null</td><td>[[[[ombCategory, ...</td><td>female</td><td>aacdab09-c4cd-406...</td><td>[[https://github....</td><td>[[[M, M, http://t...</td><td>false</td><td>null</td><td>[[Gulgowski816, [...</td><td>Patient</td><td>[[phone, home, 55...</td><td>[&lt;div xmlns=&quot;http...</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----------+--------------------+----------------+--------------------+------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+------------+--------------------+--------------------+\n",
       "|             address| birthDate|       communication|deceasedDateTime|           extension|gender|                  id|          identifier|       maritalStatus|multipleBirthBoolean|multipleBirthInteger|                name|resourceType|             telecom|                text|\n",
       "+--------------------+----------+--------------------+----------------+--------------------+------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+------------+--------------------+--------------------+\n",
       "|[[Boston, US, [[[...|1997-03-06|[[[[[en-US, Engli...|            null|[[[[ombCategory, ...|  male|d0dc1b65-77be-4a4...|[[https://github....|[[[S, Never Marri...|               false|                null|[[O'Hara248, [Ald...|     Patient|[[phone, home, 55...|[<div xmlns=\"http...|\n",
       "|[[Westport, US, [...|1999-07-23|[[[[[en-US, Engli...|            null|[[[[ombCategory, ...|  male|842c8622-913d-47c...|[[https://github....|[[[S, Never Marri...|               false|                null|[[Huels583, [Ambr...|     Patient|[[phone, home, 55...|[<div xmlns=\"http...|\n",
       "|[[Lowell, US, [[[...|1965-06-10|[[[[[zh, Chinese,...|            null|[[[[ombCategory, ...|  male|12485225-3175-4b3...|[[https://github....|[[[M, M, http://t...|               false|                null|[[Volkman526, [Be...|     Patient|[[phone, home, 55...|[<div xmlns=\"http...|\n",
       "|[[Rutland, US, [[...|2017-12-19|[[[[[hi, Hindi, h...|            null|[[[[ombCategory, ...|  male|fdf2bbc0-05c7-4ce...|[[https://github....|[[[S, Never Marri...|               false|                null|[[Hermiston71, [W...|     Patient|[[phone, home, 55...|[<div xmlns=\"http...|\n",
       "|[[Lowell, US, [[[...|1985-12-09|[[[[[en-US, Engli...|            null|[[[[ombCategory, ...|female|be2d4a5f-a868-4cc...|[[https://github....|[[[M, M, http://t...|               false|                null|[[Nader710, [Lash...|     Patient|[[phone, home, 55...|[<div xmlns=\"http...|\n",
       "+--------------------+----------+--------------------+----------------+--------------------+------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+------------+--------------------+--------------------+"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# patient = patient.limit(100000).cache()\n",
    "patient.limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- address: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- city: string (nullable = true)\n",
      " |    |    |-- country: string (nullable = true)\n",
      " |    |    |-- extension: array (nullable = true)\n",
      " |    |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |    |-- extension: array (nullable = true)\n",
      " |    |    |    |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |    |    |    |-- url: string (nullable = true)\n",
      " |    |    |    |    |    |    |-- valueDecimal: double (nullable = true)\n",
      " |    |    |    |    |-- url: string (nullable = true)\n",
      " |    |    |-- line: array (nullable = true)\n",
      " |    |    |    |-- element: string (containsNull = true)\n",
      " |    |    |-- postalCode: string (nullable = true)\n",
      " |    |    |-- state: string (nullable = true)\n",
      " |-- birthDate: string (nullable = true)\n",
      " |-- communication: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- language: struct (nullable = true)\n",
      " |    |    |    |-- coding: array (nullable = true)\n",
      " |    |    |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |    |    |-- code: string (nullable = true)\n",
      " |    |    |    |    |    |-- display: string (nullable = true)\n",
      " |    |    |    |    |    |-- system: string (nullable = true)\n",
      " |    |    |    |-- text: string (nullable = true)\n",
      " |-- deceasedDateTime: string (nullable = true)\n",
      " |-- extension: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- extension: array (nullable = true)\n",
      " |    |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |    |-- url: string (nullable = true)\n",
      " |    |    |    |    |-- valueCoding: struct (nullable = true)\n",
      " |    |    |    |    |    |-- code: string (nullable = true)\n",
      " |    |    |    |    |    |-- display: string (nullable = true)\n",
      " |    |    |    |    |    |-- system: string (nullable = true)\n",
      " |    |    |    |    |-- valueString: string (nullable = true)\n",
      " |    |    |-- url: string (nullable = true)\n",
      " |    |    |-- valueAddress: struct (nullable = true)\n",
      " |    |    |    |-- city: string (nullable = true)\n",
      " |    |    |    |-- country: string (nullable = true)\n",
      " |    |    |    |-- state: string (nullable = true)\n",
      " |    |    |-- valueCode: string (nullable = true)\n",
      " |    |    |-- valueDecimal: double (nullable = true)\n",
      " |    |    |-- valueString: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- id: string (nullable = true)\n",
      " |-- identifier: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- system: string (nullable = true)\n",
      " |    |    |-- type: struct (nullable = true)\n",
      " |    |    |    |-- coding: array (nullable = true)\n",
      " |    |    |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |    |    |-- code: string (nullable = true)\n",
      " |    |    |    |    |    |-- display: string (nullable = true)\n",
      " |    |    |    |    |    |-- system: string (nullable = true)\n",
      " |    |    |    |-- text: string (nullable = true)\n",
      " |    |    |-- value: string (nullable = true)\n",
      " |-- maritalStatus: struct (nullable = true)\n",
      " |    |-- coding: array (nullable = true)\n",
      " |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |-- code: string (nullable = true)\n",
      " |    |    |    |-- display: string (nullable = true)\n",
      " |    |    |    |-- system: string (nullable = true)\n",
      " |    |-- text: string (nullable = true)\n",
      " |-- multipleBirthBoolean: boolean (nullable = true)\n",
      " |-- multipleBirthInteger: long (nullable = true)\n",
      " |-- name: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- family: string (nullable = true)\n",
      " |    |    |-- given: array (nullable = true)\n",
      " |    |    |    |-- element: string (containsNull = true)\n",
      " |    |    |-- prefix: array (nullable = true)\n",
      " |    |    |    |-- element: string (containsNull = true)\n",
      " |    |    |-- suffix: array (nullable = true)\n",
      " |    |    |    |-- element: string (containsNull = true)\n",
      " |    |    |-- use: string (nullable = true)\n",
      " |-- resourceType: string (nullable = true)\n",
      " |-- telecom: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- system: string (nullable = true)\n",
      " |    |    |-- use: string (nullable = true)\n",
      " |    |    |-- value: string (nullable = true)\n",
      " |-- text: struct (nullable = true)\n",
      " |    |-- div: string (nullable = true)\n",
      " |    |-- status: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "patient.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='section_5'></a>\n",
    "## Section 5.  Link Condition with Patients (diabetes)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "patient_all = patient.select(col('id').alias('patient_id'), col('birthDate').alias('birth_date'), 'gender')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>birth_date</th><th>gender</th></tr>\n",
       "<tr><td>2d913c00-48ad-4af...</td><td>2004-05-23</td><td>male</td></tr>\n",
       "<tr><td>2ecfa822-e49b-420...</td><td>1998-11-22</td><td>male</td></tr>\n",
       "<tr><td>4a9668b1-ff30-495...</td><td>2015-02-20</td><td>male</td></tr>\n",
       "<tr><td>39b933da-1358-429...</td><td>1971-04-27</td><td>male</td></tr>\n",
       "<tr><td>a60f42cd-76c3-42b...</td><td>1982-04-19</td><td>male</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----------+------+\n",
       "|          patient_id|birth_date|gender|\n",
       "+--------------------+----------+------+\n",
       "|2d913c00-48ad-4af...|2004-05-23|  male|\n",
       "|2ecfa822-e49b-420...|1998-11-22|  male|\n",
       "|4a9668b1-ff30-495...|2015-02-20|  male|\n",
       "|39b933da-1358-429...|1971-04-27|  male|\n",
       "|a60f42cd-76c3-42b...|1982-04-19|  male|\n",
       "+--------------------+----------+------+"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "patient_all.limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8081"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Number of records with Diabetes\n",
    "condition_diabetes.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "117311"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Number of patients\n",
    "patient_all.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 4 ms, sys: 0 ns, total: 4 ms\n",
      "Wall time: 22.3 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "8081"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# patient_diabetes = patient_all.join(condition_diabetes, (patient_all.patient_id == condition_diabetes.patient_id), how='inner')\n",
    "\n",
    "# To prevent duplicate column\n",
    "%time patient_diabetes = patient_all.join(condition_diabetes, ['patient_id'], how='inner')\n",
    "\n",
    "# patient_diabetes = patient_diabetes.limit(100000).cache()\n",
    "patient_diabetes.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>birth_date</th><th>gender</th><th>condition_status</th><th>verification_status</th><th>snomed_code</th><th>snomed_name</th><th>first_observation_date</th></tr>\n",
       "<tr><td>08b1c9ae-9d60-4f5...</td><td>1966-03-10</td><td>female</td><td>active</td><td>confirmed</td><td>44054006</td><td>Diabetes</td><td>1989-07-13</td></tr>\n",
       "<tr><td>09c21f37-89b6-41e...</td><td>1948-11-01</td><td>female</td><td>active</td><td>confirmed</td><td>44054006</td><td>Diabetes</td><td>1992-05-11</td></tr>\n",
       "<tr><td>10209746-e589-450...</td><td>1982-02-02</td><td>male</td><td>active</td><td>confirmed</td><td>44054006</td><td>Diabetes</td><td>2010-04-13</td></tr>\n",
       "<tr><td>10b99cb0-5a8e-478...</td><td>1948-08-10</td><td>male</td><td>active</td><td>confirmed</td><td>44054006</td><td>Diabetes</td><td>1977-06-14</td></tr>\n",
       "<tr><td>1170f625-d812-438...</td><td>1964-05-23</td><td>male</td><td>active</td><td>confirmed</td><td>44054006</td><td>Diabetes</td><td>1992-09-26</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----------+------+----------------+-------------------+-----------+-----------+----------------------+\n",
       "|          patient_id|birth_date|gender|condition_status|verification_status|snomed_code|snomed_name|first_observation_date|\n",
       "+--------------------+----------+------+----------------+-------------------+-----------+-----------+----------------------+\n",
       "|08b1c9ae-9d60-4f5...|1966-03-10|female|          active|          confirmed|   44054006|   Diabetes|            1989-07-13|\n",
       "|09c21f37-89b6-41e...|1948-11-01|female|          active|          confirmed|   44054006|   Diabetes|            1992-05-11|\n",
       "|10209746-e589-450...|1982-02-02|  male|          active|          confirmed|   44054006|   Diabetes|            2010-04-13|\n",
       "|10b99cb0-5a8e-478...|1948-08-10|  male|          active|          confirmed|   44054006|   Diabetes|            1977-06-14|\n",
       "|1170f625-d812-438...|1964-05-23|  male|          active|          confirmed|   44054006|   Diabetes|            1992-09-26|\n",
       "+--------------------+----------+------+----------------+-------------------+-----------+-----------+----------------------+"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "patient_diabetes.limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- patient_id: string (nullable = true)\n",
      " |-- birth_date: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- condition_status: string (nullable = true)\n",
      " |-- verification_status: string (nullable = true)\n",
      " |-- snomed_code: string (nullable = true)\n",
      " |-- snomed_name: string (nullable = true)\n",
      " |-- first_observation_date: date (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "patient_diabetes.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='section_6'></a>\n",
    "## Section 6.  Create Longitudinal Patient Record"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Prepare condition resource"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>condition_status</th><th>verification_status</th><th>snomed_code</th><th>snomed_name</th><th>first_observation_date</th></tr>\n",
       "<tr><td>ebac0cc1-07a5-449...</td><td>resolved</td><td>confirmed</td><td>43878008</td><td>Streptococcal sor...</td><td>2009-06-30</td></tr>\n",
       "<tr><td>ecd26f59-fe71-4a8...</td><td>resolved</td><td>confirmed</td><td>10509002</td><td>Acute bronchitis ...</td><td>2009-08-20</td></tr>\n",
       "<tr><td>ebac0cc1-07a5-449...</td><td>resolved</td><td>confirmed</td><td>195662009</td><td>Acute viral phary...</td><td>2010-06-09</td></tr>\n",
       "<tr><td>ebac0cc1-07a5-449...</td><td>resolved</td><td>confirmed</td><td>444814009</td><td>Viral sinusitis (...</td><td>2011-05-12</td></tr>\n",
       "<tr><td>ecd26f59-fe71-4a8...</td><td>resolved</td><td>confirmed</td><td>58150001</td><td>Fracture of clavicle</td><td>2010-03-05</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----------------+-------------------+-----------+--------------------+----------------------+\n",
       "|          patient_id|condition_status|verification_status|snomed_code|         snomed_name|first_observation_date|\n",
       "+--------------------+----------------+-------------------+-----------+--------------------+----------------------+\n",
       "|a60f42cd-76c3-42b...|          active|          confirmed|  162864005|Body mass index 3...|            1993-08-23|\n",
       "|f03ce2e7-e7e4-43b...|          active|          confirmed|   15777000|         Prediabetes|            1985-08-20|\n",
       "|39b933da-1358-429...|          active|          confirmed|   38341003|        Hypertension|            1989-06-20|\n",
       "|2771f421-587e-4d6...|          active|          confirmed|  162864005|Body mass index 3...|            2001-07-09|\n",
       "|2d913c00-48ad-4af...|        resolved|          confirmed|  263102004|Fracture subluxat...|            2009-12-23|\n",
       "+--------------------+----------------+-------------------+-----------+--------------------+----------------------+"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "condition_all = condition.\\\n",
    "withColumn('condition_status', condition['clinicalStatus.coding.code'].getItem(0)).\\\n",
    "withColumn('verification_status', condition['verificationStatus.coding.code'].getItem(0)).\\\n",
    "withColumn('snomed_code', condition['code.coding.code'].getItem(0)).\\\n",
    "withColumn('snomed_name', condition['code.coding.display'].getItem(0)).\\\n",
    "select(\\\n",
    "       (condition['subject.reference'].substr(10, 40).alias('patient_id')),\n",
    "       'condition_status',\\\n",
    "       'verification_status',\\\n",
    "       'snomed_code', \\\n",
    "       'snomed_name',\\\n",
    "       to_date(condition['onsetDateTime']).alias('first_observation_date'))\n",
    "\n",
    "condition_all.limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>target_code</th><th>target_name</th><th>target_first_date</th></tr>\n",
       "<tr><td>89c668fd-0e66-494...</td><td>44054006</td><td>Diabetes</td><td>2013-11-26</td></tr>\n",
       "<tr><td>71d5b64e-4ea0-451...</td><td>44054006</td><td>Diabetes</td><td>1994-12-31</td></tr>\n",
       "<tr><td>82e4efba-ab48-428...</td><td>44054006</td><td>Diabetes</td><td>2015-12-26</td></tr>\n",
       "<tr><td>2a9048fc-56d9-489...</td><td>44054006</td><td>Diabetes</td><td>1982-03-29</td></tr>\n",
       "<tr><td>2c945863-0afc-46b...</td><td>44054006</td><td>Diabetes</td><td>1982-09-08</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+-----------+-----------+-----------------+\n",
       "|          patient_id|target_code|target_name|target_first_date|\n",
       "+--------------------+-----------+-----------+-----------------+\n",
       "|1826f710-e916-4e3...|   44054006|   Diabetes|       1988-11-11|\n",
       "|d1b9f07e-cc69-4f9...|   44054006|   Diabetes|       2007-08-06|\n",
       "|3546dd2e-0214-47f...|   44054006|   Diabetes|       2001-12-16|\n",
       "|01d846aa-86b8-409...|   44054006|   Diabetes|       1998-11-27|\n",
       "|dd005704-76e5-477...|   44054006|   Diabetes|       2006-05-17|\n",
       "+--------------------+-----------+-----------+-----------------+"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "condition_target = condition_diabetes.\\\n",
    "    withColumnRenamed('snomed_code', 'target_code').\\\n",
    "    withColumnRenamed('snomed_name', 'target_name').\\\n",
    "    withColumnRenamed('first_observation_date', 'target_first_date').\\\n",
    "    drop('condition_status', 'verification_status')\n",
    "\n",
    "condition_target.limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Get all other conditions for patients with diabetes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 4 ms, sys: 0 ns, total: 4 ms\n",
      "Wall time: 10.7 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "120835"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%time patient_condition = condition_all.join(condition_target, ['patient_id'], how='inner')\n",
    "\n",
    "patient_condition.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>condition_status</th><th>verification_status</th><th>snomed_code</th><th>snomed_name</th><th>first_observation_date</th><th>target_code</th><th>target_name</th><th>target_first_date</th></tr>\n",
       "<tr><td>08b1c9ae-9d60-4f5...</td><td>active</td><td>confirmed</td><td>38341003</td><td>Hypertension</td><td>1989-07-13</td><td>44054006</td><td>Diabetes</td><td>1989-07-13</td></tr>\n",
       "<tr><td>08b1c9ae-9d60-4f5...</td><td>active</td><td>confirmed</td><td>44054006</td><td>Diabetes</td><td>1989-07-13</td><td>44054006</td><td>Diabetes</td><td>1989-07-13</td></tr>\n",
       "<tr><td>08b1c9ae-9d60-4f5...</td><td>active</td><td>confirmed</td><td>302870006</td><td>Hypertriglyceride...</td><td>1992-07-16</td><td>44054006</td><td>Diabetes</td><td>1989-07-13</td></tr>\n",
       "<tr><td>08b1c9ae-9d60-4f5...</td><td>active</td><td>confirmed</td><td>431855005</td><td>Chronic kidney di...</td><td>1992-07-16</td><td>44054006</td><td>Diabetes</td><td>1989-07-13</td></tr>\n",
       "<tr><td>08b1c9ae-9d60-4f5...</td><td>active</td><td>confirmed</td><td>127013003</td><td>Diabetic renal di...</td><td>1992-07-16</td><td>44054006</td><td>Diabetes</td><td>1989-07-13</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----------------+-------------------+-----------+--------------------+----------------------+-----------+-----------+-----------------+\n",
       "|          patient_id|condition_status|verification_status|snomed_code|         snomed_name|first_observation_date|target_code|target_name|target_first_date|\n",
       "+--------------------+----------------+-------------------+-----------+--------------------+----------------------+-----------+-----------+-----------------+\n",
       "|12829da0-007b-4ad...|          active|          confirmed|   15777000|         Prediabetes|            1983-04-20|   44054006|   Diabetes|       1986-04-23|\n",
       "|12829da0-007b-4ad...|          active|          confirmed|   38341003|        Hypertension|            1986-04-23|   44054006|   Diabetes|       1986-04-23|\n",
       "|12829da0-007b-4ad...|          active|          confirmed|   44054006|            Diabetes|            1986-04-23|   44054006|   Diabetes|       1986-04-23|\n",
       "|12829da0-007b-4ad...|          active|          confirmed|  302870006|Hypertriglyceride...|            1988-11-02|   44054006|   Diabetes|       1986-04-23|\n",
       "|12829da0-007b-4ad...|          active|          confirmed|  237602007|Metabolic syndrom...|            1988-11-02|   44054006|   Diabetes|       1986-04-23|\n",
       "+--------------------+----------------+-------------------+-----------+--------------------+----------------------+-----------+-----------+-----------------+"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# patient_condition = patient_condition.limit(100000).cache()\n",
    "patient_condition.limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Restrict to include only relevant comorbities and complications"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>condition_status</th><th>verification_status</th><th>snomed_code</th><th>snomed_name</th><th>first_observation_date</th><th>target_code</th><th>target_name</th><th>target_first_date</th></tr>\n",
       "<tr><td>08b1c9ae-9d60-4f5...</td><td>active</td><td>confirmed</td><td>38341003</td><td>Hypertension</td><td>1989-07-13</td><td>44054006</td><td>Diabetes</td><td>1989-07-13</td></tr>\n",
       "<tr><td>08b1c9ae-9d60-4f5...</td><td>active</td><td>confirmed</td><td>127013003</td><td>Diabetic renal di...</td><td>1992-07-16</td><td>44054006</td><td>Diabetes</td><td>1989-07-13</td></tr>\n",
       "<tr><td>08b1c9ae-9d60-4f5...</td><td>active</td><td>confirmed</td><td>90781000119102</td><td>Microalbuminuria ...</td><td>2004-08-05</td><td>44054006</td><td>Diabetes</td><td>1989-07-13</td></tr>\n",
       "<tr><td>09c21f37-89b6-41e...</td><td>active</td><td>confirmed</td><td>38341003</td><td>Hypertension</td><td>1966-12-26</td><td>44054006</td><td>Diabetes</td><td>1992-05-11</td></tr>\n",
       "<tr><td>09c21f37-89b6-41e...</td><td>active</td><td>confirmed</td><td>127013003</td><td>Diabetic renal di...</td><td>1996-05-20</td><td>44054006</td><td>Diabetes</td><td>1992-05-11</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----------------+-------------------+--------------+--------------------+----------------------+-----------+-----------+-----------------+\n",
       "|          patient_id|condition_status|verification_status|   snomed_code|         snomed_name|first_observation_date|target_code|target_name|target_first_date|\n",
       "+--------------------+----------------+-------------------+--------------+--------------------+----------------------+-----------+-----------+-----------------+\n",
       "|08b1c9ae-9d60-4f5...|          active|          confirmed|      38341003|        Hypertension|            1989-07-13|   44054006|   Diabetes|       1989-07-13|\n",
       "|08b1c9ae-9d60-4f5...|          active|          confirmed|     127013003|Diabetic renal di...|            1992-07-16|   44054006|   Diabetes|       1989-07-13|\n",
       "|08b1c9ae-9d60-4f5...|          active|          confirmed|90781000119102|Microalbuminuria ...|            2004-08-05|   44054006|   Diabetes|       1989-07-13|\n",
       "|09c21f37-89b6-41e...|          active|          confirmed|      38341003|        Hypertension|            1966-12-26|   44054006|   Diabetes|       1992-05-11|\n",
       "|09c21f37-89b6-41e...|          active|          confirmed|     127013003|Diabetic renal di...|            1996-05-20|   44054006|   Diabetes|       1992-05-11|\n",
       "+--------------------+----------------+-------------------+--------------+--------------------+----------------------+-----------+-----------+-----------------+"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Limit to the relevant status\n",
    "# patient_condition = patient_condition.filter((patient_condition['condition_status'] == 'active') & (patient_condition['verification_status'] == 'confirmed'))\n",
    "\n",
    "# Limit to the relevant comorbities and complications\n",
    "patient_relevant = patient_condition.where(col('snomed_code').isin({\\\n",
    "        '53741008',\\\n",
    "        '709044004',\\\n",
    "        '49436004',\\\n",
    "        '230690007',\\\n",
    "        '38341003',\\\n",
    "        '84114007',\\\n",
    "        '400047006',\\\n",
    "        '69896004',\\\n",
    "        '86049000',\\\n",
    "        '363346000',\\\n",
    "        '64859006',\\\n",
    "        '35489007',\\\n",
    "        '195967001',\\\n",
    "        '13645005',\\\n",
    "        '52448006',\\\n",
    "        '391193001',\\\n",
    "        '84757009',\\\n",
    "        '40930008',\\\n",
    "        '1855002',\\\n",
    "        '56265001',\\\n",
    "        '230690007',\\\n",
    "        '65956007',\\\n",
    "        '60951000119105',\\\n",
    "        '105597003',\\\n",
    "        '97331000119101',\\\n",
    "        '422034002',\\\n",
    "        '1501000119109',\\\n",
    "        '1551000119108',\\\n",
    "        '90708001',\\\n",
    "        '127013003',\\\n",
    "        '46177005',\\\n",
    "        '57182000',\\\n",
    "        '386033004',\\\n",
    "        '368581000119106',\\\n",
    "        '81723002 ',\\\n",
    "        '161622006',\\\n",
    "        '429280009',\\\n",
    "        '161621004',\\\n",
    "        '698423002',\\\n",
    "        '90781000119102',\\\n",
    "        '157141000119108'\n",
    "            }))\n",
    "\n",
    "patient_relevant.limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 8 ms, sys: 4 ms, total: 12 ms\n",
      "Wall time: 39.2 ms\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>target_code</th><th>target_name</th><th>snomed_code</th><th>snomed_name</th><th>target_first_date</th><th>first_observation_date</th><th>last_observation_date</th></tr>\n",
       "<tr><td>123a02a9-5d5a-402...</td><td>44054006</td><td>Diabetes</td><td>38341003</td><td>Hypertension</td><td>2009-10-17</td><td>1997-10-18</td><td>1997-10-18</td></tr>\n",
       "<tr><td>123a02a9-5d5a-402...</td><td>44054006</td><td>Diabetes</td><td>368581000119106</td><td>Neuropathy due to...</td><td>2009-10-17</td><td>2019-01-05</td><td>2019-01-05</td></tr>\n",
       "<tr><td>1cf894aa-e622-4be...</td><td>44054006</td><td>Diabetes</td><td>53741008</td><td>Coronary Heart Di...</td><td>1993-12-28</td><td>2013-03-26</td><td>2013-03-26</td></tr>\n",
       "<tr><td>1cf894aa-e622-4be...</td><td>44054006</td><td>Diabetes</td><td>422034002</td><td>Diabetic retinopa...</td><td>1993-12-28</td><td>2014-11-18</td><td>2014-11-18</td></tr>\n",
       "<tr><td>22b284c3-5200-493...</td><td>44054006</td><td>Diabetes</td><td>422034002</td><td>Diabetic retinopa...</td><td>1940-03-08</td><td>1940-03-08</td><td>1940-03-08</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+-----------+-----------+--------------+--------------------+-----------------+----------------------+---------------------+\n",
       "|          patient_id|target_code|target_name|   snomed_code|         snomed_name|target_first_date|first_observation_date|last_observation_date|\n",
       "+--------------------+-----------+-----------+--------------+--------------------+-----------------+----------------------+---------------------+\n",
       "|08b1c9ae-9d60-4f5...|   44054006|   Diabetes|      38341003|        Hypertension|       1989-07-13|            1989-07-13|           1989-07-13|\n",
       "|08b1c9ae-9d60-4f5...|   44054006|   Diabetes|     127013003|Diabetic renal di...|       1989-07-13|            1992-07-16|           1992-07-16|\n",
       "|08b1c9ae-9d60-4f5...|   44054006|   Diabetes|90781000119102|Microalbuminuria ...|       1989-07-13|            2004-08-05|           2004-08-05|\n",
       "|09c21f37-89b6-41e...|   44054006|   Diabetes|      38341003|        Hypertension|       1992-05-11|            1966-12-26|           1966-12-26|\n",
       "|09c21f37-89b6-41e...|   44054006|   Diabetes|     127013003|Diabetic renal di...|       1992-05-11|            1996-05-20|           1996-05-20|\n",
       "+--------------------+-----------+-----------+--------------+--------------------+-----------------+----------------------+---------------------+"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Identify the first / last date of each condition\n",
    "%time lpr_group = patient_relevant.groupby('patient_id', 'target_code', 'target_name', 'snomed_code', 'snomed_name').\\\n",
    "    agg(\\\n",
    "        min('target_first_date').alias('target_first_date'),\\\n",
    "        min('first_observation_date').alias('first_observation_date'),\\\n",
    "        max('first_observation_date').alias('last_observation_date'),\\\n",
    "       )\n",
    "\n",
    "lpr_group.limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Add patient birth data and gender.  We now have row-based LPR"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>birth_date</th><th>gender</th><th>target_code</th><th>target_name</th><th>snomed_code</th><th>snomed_name</th><th>target_first_date</th><th>first_observation_date</th><th>last_observation_date</th></tr>\n",
       "<tr><td>08b1c9ae-9d60-4f5...</td><td>1966-03-10</td><td>female</td><td>44054006</td><td>Diabetes</td><td>38341003</td><td>Hypertension</td><td>1989-07-13</td><td>1989-07-13</td><td>1989-07-13</td></tr>\n",
       "<tr><td>08b1c9ae-9d60-4f5...</td><td>1966-03-10</td><td>female</td><td>44054006</td><td>Diabetes</td><td>127013003</td><td>Diabetic renal di...</td><td>1989-07-13</td><td>1992-07-16</td><td>1992-07-16</td></tr>\n",
       "<tr><td>08b1c9ae-9d60-4f5...</td><td>1966-03-10</td><td>female</td><td>44054006</td><td>Diabetes</td><td>90781000119102</td><td>Microalbuminuria ...</td><td>1989-07-13</td><td>2004-08-05</td><td>2004-08-05</td></tr>\n",
       "<tr><td>09c21f37-89b6-41e...</td><td>1948-11-01</td><td>female</td><td>44054006</td><td>Diabetes</td><td>38341003</td><td>Hypertension</td><td>1992-05-11</td><td>1966-12-26</td><td>1966-12-26</td></tr>\n",
       "<tr><td>09c21f37-89b6-41e...</td><td>1948-11-01</td><td>female</td><td>44054006</td><td>Diabetes</td><td>127013003</td><td>Diabetic renal di...</td><td>1992-05-11</td><td>1996-05-20</td><td>1996-05-20</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----------+------+-----------+-----------+--------------+--------------------+-----------------+----------------------+---------------------+\n",
       "|          patient_id|birth_date|gender|target_code|target_name|   snomed_code|         snomed_name|target_first_date|first_observation_date|last_observation_date|\n",
       "+--------------------+----------+------+-----------+-----------+--------------+--------------------+-----------------+----------------------+---------------------+\n",
       "|08b1c9ae-9d60-4f5...|1966-03-10|female|   44054006|   Diabetes|      38341003|        Hypertension|       1989-07-13|            1989-07-13|           1989-07-13|\n",
       "|08b1c9ae-9d60-4f5...|1966-03-10|female|   44054006|   Diabetes|     127013003|Diabetic renal di...|       1989-07-13|            1992-07-16|           1992-07-16|\n",
       "|08b1c9ae-9d60-4f5...|1966-03-10|female|   44054006|   Diabetes|90781000119102|Microalbuminuria ...|       1989-07-13|            2004-08-05|           2004-08-05|\n",
       "|09c21f37-89b6-41e...|1948-11-01|female|   44054006|   Diabetes|      38341003|        Hypertension|       1992-05-11|            1966-12-26|           1966-12-26|\n",
       "|09c21f37-89b6-41e...|1948-11-01|female|   44054006|   Diabetes|     127013003|Diabetic renal di...|       1992-05-11|            1996-05-20|           1996-05-20|\n",
       "+--------------------+----------+------+-----------+-----------+--------------+--------------------+-----------------+----------------------+---------------------+"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lpr_row = patient_all.join(lpr_group, ['patient_id'], how='inner')\n",
    "\n",
    "lpr_row.limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "21982"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lpr_row.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "21982"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# lpr_row.limit(100000).cache()\n",
    "lpr_row.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Convert row-based LPR into column-based LPR"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 84 ms, sys: 52 ms, total: 136 ms\n",
      "Wall time: 11.5 s\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>patient_id</th><th>birth_date</th><th>gender</th><th>target_code</th><th>target_name</th><th>target_first_date</th><th>127013003_first_dt</th><th>127013003_last_dt</th><th>1501000119109_first_dt</th><th>1501000119109_last_dt</th><th>1551000119108_first_dt</th><th>1551000119108_last_dt</th><th>157141000119108_first_dt</th><th>157141000119108_last_dt</th><th>161621004_first_dt</th><th>161621004_last_dt</th><th>161622006_first_dt</th><th>161622006_last_dt</th><th>195967001_first_dt</th><th>195967001_last_dt</th><th>230690007_first_dt</th><th>230690007_last_dt</th><th>368581000119106_first_dt</th><th>368581000119106_last_dt</th><th>38341003_first_dt</th><th>38341003_last_dt</th><th>422034002_first_dt</th><th>422034002_last_dt</th><th>429280009_first_dt</th><th>429280009_last_dt</th><th>46177005_first_dt</th><th>46177005_last_dt</th><th>49436004_first_dt</th><th>49436004_last_dt</th><th>53741008_first_dt</th><th>53741008_last_dt</th><th>60951000119105_first_dt</th><th>60951000119105_last_dt</th><th>64859006_first_dt</th><th>64859006_last_dt</th><th>698423002_first_dt</th><th>698423002_last_dt</th><th>69896004_first_dt</th><th>69896004_last_dt</th><th>84757009_first_dt</th><th>84757009_last_dt</th><th>90781000119102_first_dt</th><th>90781000119102_last_dt</th><th>97331000119101_first_dt</th><th>97331000119101_last_dt</th></tr>\n",
       "<tr><td>08b1c9ae-9d60-4f5...</td><td>1966-03-10</td><td>female</td><td>44054006</td><td>Diabetes</td><td>1989-07-13</td><td>1992-07-16</td><td>1992-07-16</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>1989-07-13</td><td>1989-07-13</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>2004-08-05</td><td>2004-08-05</td><td>null</td><td>null</td></tr>\n",
       "<tr><td>09c21f37-89b6-41e...</td><td>1948-11-01</td><td>female</td><td>44054006</td><td>Diabetes</td><td>1992-05-11</td><td>1996-05-20</td><td>1996-05-20</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>1996-05-20</td><td>1996-05-20</td><td>1966-12-26</td><td>1966-12-26</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>2008-11-03</td><td>2008-11-03</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td></tr>\n",
       "<tr><td>10209746-e589-450...</td><td>1982-02-02</td><td>male</td><td>44054006</td><td>Diabetes</td><td>2010-04-13</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>2000-03-28</td><td>2000-03-28</td><td>2010-04-13</td><td>2010-04-13</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td></tr>\n",
       "<tr><td>10b99cb0-5a8e-478...</td><td>1948-08-10</td><td>male</td><td>44054006</td><td>Diabetes</td><td>1977-06-14</td><td>1995-05-16</td><td>1995-05-16</td><td>null</td><td>null</td><td>1996-07-09</td><td>1996-07-09</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>1977-06-14</td><td>1977-06-14</td><td>1992-06-23</td><td>1992-06-23</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td></tr>\n",
       "<tr><td>1170f625-d812-438...</td><td>1964-05-23</td><td>male</td><td>44054006</td><td>Diabetes</td><td>1992-09-26</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>1992-09-26</td><td>1992-09-26</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>null</td><td>1980-05-19</td><td>1980-05-19</td><td>null</td><td>null</td><td>null</td><td>null</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----------+------+-----------+-----------+-----------------+------------------+-----------------+----------------------+---------------------+----------------------+---------------------+------------------------+-----------------------+------------------+-----------------+------------------+-----------------+------------------+-----------------+------------------+-----------------+------------------------+-----------------------+-----------------+----------------+------------------+-----------------+------------------+-----------------+-----------------+----------------+-----------------+----------------+-----------------+----------------+-----------------------+----------------------+-----------------+----------------+------------------+-----------------+-----------------+----------------+-----------------+----------------+-----------------------+----------------------+-----------------------+----------------------+\n",
       "|          patient_id|birth_date|gender|target_code|target_name|target_first_date|127013003_first_dt|127013003_last_dt|1501000119109_first_dt|1501000119109_last_dt|1551000119108_first_dt|1551000119108_last_dt|157141000119108_first_dt|157141000119108_last_dt|161621004_first_dt|161621004_last_dt|161622006_first_dt|161622006_last_dt|195967001_first_dt|195967001_last_dt|230690007_first_dt|230690007_last_dt|368581000119106_first_dt|368581000119106_last_dt|38341003_first_dt|38341003_last_dt|422034002_first_dt|422034002_last_dt|429280009_first_dt|429280009_last_dt|46177005_first_dt|46177005_last_dt|49436004_first_dt|49436004_last_dt|53741008_first_dt|53741008_last_dt|60951000119105_first_dt|60951000119105_last_dt|64859006_first_dt|64859006_last_dt|698423002_first_dt|698423002_last_dt|69896004_first_dt|69896004_last_dt|84757009_first_dt|84757009_last_dt|90781000119102_first_dt|90781000119102_last_dt|97331000119101_first_dt|97331000119101_last_dt|\n",
       "+--------------------+----------+------+-----------+-----------+-----------------+------------------+-----------------+----------------------+---------------------+----------------------+---------------------+------------------------+-----------------------+------------------+-----------------+------------------+-----------------+------------------+-----------------+------------------+-----------------+------------------------+-----------------------+-----------------+----------------+------------------+-----------------+------------------+-----------------+-----------------+----------------+-----------------+----------------+-----------------+----------------+-----------------------+----------------------+-----------------+----------------+------------------+-----------------+-----------------+----------------+-----------------+----------------+-----------------------+----------------------+-----------------------+----------------------+\n",
       "|08b1c9ae-9d60-4f5...|1966-03-10|female|   44054006|   Diabetes|       1989-07-13|        1992-07-16|       1992-07-16|                  null|                 null|                  null|                 null|                    null|                   null|              null|             null|              null|             null|              null|             null|              null|             null|                    null|                   null|       1989-07-13|      1989-07-13|              null|             null|              null|             null|             null|            null|             null|            null|             null|            null|                   null|                  null|             null|            null|              null|             null|             null|            null|             null|            null|             2004-08-05|            2004-08-05|                   null|                  null|\n",
       "|09c21f37-89b6-41e...|1948-11-01|female|   44054006|   Diabetes|       1992-05-11|        1996-05-20|       1996-05-20|                  null|                 null|                  null|                 null|                    null|                   null|              null|             null|              null|             null|              null|             null|              null|             null|              1996-05-20|             1996-05-20|       1966-12-26|      1966-12-26|              null|             null|              null|             null|             null|            null|             null|            null|             null|            null|                   null|                  null|       2008-11-03|      2008-11-03|              null|             null|             null|            null|             null|            null|                   null|                  null|                   null|                  null|\n",
       "|10209746-e589-450...|1982-02-02|  male|   44054006|   Diabetes|       2010-04-13|              null|             null|                  null|                 null|                  null|                 null|                    null|                   null|              null|             null|              null|             null|              null|             null|              null|             null|                    null|                   null|       2000-03-28|      2000-03-28|        2010-04-13|       2010-04-13|              null|             null|             null|            null|             null|            null|             null|            null|                   null|                  null|             null|            null|              null|             null|             null|            null|             null|            null|                   null|                  null|                   null|                  null|\n",
       "|10b99cb0-5a8e-478...|1948-08-10|  male|   44054006|   Diabetes|       1977-06-14|        1995-05-16|       1995-05-16|                  null|                 null|            1996-07-09|           1996-07-09|                    null|                   null|              null|             null|              null|             null|              null|             null|              null|             null|                    null|                   null|       1977-06-14|      1977-06-14|        1992-06-23|       1992-06-23|              null|             null|             null|            null|             null|            null|             null|            null|                   null|                  null|             null|            null|              null|             null|             null|            null|             null|            null|                   null|                  null|                   null|                  null|\n",
       "|1170f625-d812-438...|1964-05-23|  male|   44054006|   Diabetes|       1992-09-26|              null|             null|                  null|                 null|                  null|                 null|                    null|                   null|              null|             null|              null|             null|              null|             null|              null|             null|                    null|                   null|       1992-09-26|      1992-09-26|              null|             null|              null|             null|             null|            null|             null|            null|             null|            null|                   null|                  null|             null|            null|              null|             null|             null|            null|       1980-05-19|      1980-05-19|                   null|                  null|                   null|                  null|\n",
       "+--------------------+----------+------+-----------+-----------+-----------------+------------------+-----------------+----------------------+---------------------+----------------------+---------------------+------------------------+-----------------------+------------------+-----------------+------------------+-----------------+------------------+-----------------+------------------+-----------------+------------------------+-----------------------+-----------------+----------------+------------------+-----------------+------------------+-----------------+-----------------+----------------+-----------------+----------------+-----------------+----------------+-----------------------+----------------------+-----------------+----------------+------------------+-----------------+-----------------+----------------+-----------------+----------------+-----------------------+----------------------+-----------------------+----------------------+"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%time lpr_col = lpr_row.groupby('patient_id', 'birth_date', 'gender', 'target_code', 'target_name', 'target_first_date').\\\n",
    "    pivot('snomed_code').\\\n",
    "    agg(\\\n",
    "        first('first_observation_date').alias('first_dt'),\\\n",
    "        last('first_observation_date').alias('last_dt')\\\n",
    "           )\n",
    "\n",
    "lpr_col.limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "7393"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lpr_col.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='section_7'></a>\n",
    "## Section 7.  Save LPR into COS"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Set-up COS credentials with write persmissions  \n",
    "Participants must provide their own COS credentials with write permissions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Save patients with diabetes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "# output_file = '/lpr/patient_diabetes'\n",
    "# mode = 'overwrite'\n",
    "\n",
    "# %time patient_diabetes.write.mode(\"overwrite\").parquet(spark_cos.url(output_file, output_bucket))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Save observations with diabetes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "# output_file = '/lpr/observation_diabetes'\n",
    "# mode = 'overwrite'\n",
    "\n",
    "# %time observation_diabetes.write.mode(\"overwrite\").parquet(spark_cos.url(output_file, output_bucket))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Save row-based LPR"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [],
   "source": [
    "# output_file = '/lpr/lpr_row'\n",
    "# mode = 'overwrite'\n",
    "\n",
    "# %time lpr_row.write.mode(\"overwrite\").parquet(spark_cos.url(output_file, output_bucket))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Save column-based LPR"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "# output_file = '/lpr/lpr_col'\n",
    "# mode = 'overwrite'\n",
    "\n",
    "# %time lpr_col.write.mode(\"overwrite\").parquet(spark_cos.url(output_file, output_bucket))"
   ]
  },
  {
   "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": []
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Sat, 14 November 2020 09:50:23'"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import datetime\n",
    "import pytz\n",
    "\n",
    "datetime.datetime.now(pytz.timezone('US/Central')).strftime(\"%a, %d %B %Y %H:%M:%S\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.7 with Spark",
   "language": "python3",
   "name": "python37"
  },
  "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.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}