PySpark_Data_Wrangling.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "e25f6362",
   "metadata": {},
   "source": [
    "# PySpark Data Wrangling"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "b8a88eb2",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import pandas as pd\n",
    "from PatientJourney import PatientJourney\n",
    "from ConstantsNamespace import PatientJourneyConstantsNamespace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "1f7a50e4",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession, DataFrame\n",
    "from functools import reduce\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import IntegerType, StringType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "2e8c3a9b",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/07/11 14:06:25 WARN Utils: Your hostname, niranjan-Virtual-Machine resolves to a loopback address: 127.0.1.1; using 172.20.253.131 instead (on interface eth0)\n",
      "23/07/11 14:06:25 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "23/07/11 14:06:27 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "23/07/11 14:06:29 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n"
     ]
    }
   ],
   "source": [
    "spark = SparkSession.builder.appName('PatientJourney').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "a9873997",
   "metadata": {},
   "outputs": [],
   "source": [
    "constants = PatientJourneyConstantsNamespace()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "890a2b47",
   "metadata": {},
   "source": [
    "### Read consumption data from separate files and union them"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "56985dbc",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 0:>                                                          (0 + 1) / 1]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "diagbh2021_df = spark.read.csv(path=constants.DIAGBH2021_FILEPATH, sep=r'\\t', header=True)\n",
    "diagmd2021_df = spark.read.csv(path=constants.DIAGMD2021_FILEPATH, sep=r'\\t', header=True)\n",
    "procmd2021_df = spark.read.csv(path=constants.PROCMD2021_FILEPATH, sep=r'\\t', header=True)\n",
    "rx2021_df     = spark.read.csv(path=constants.RX2021_FILEPATH, sep=r'\\t', header=True)\n",
    "servbh2021_df = spark.read.csv(path=constants.SERVBH2021_FILEPATH, sep=r'\\t', header=True)\n",
    "servmd2021_df = spark.read.csv(path=constants.SERVMD2021_FILEPATH, sep=r'\\t', header=True)\n",
    "\n",
    "diagbh2022_df = spark.read.csv(path=constants.DIAGBH2022_FILEPATH, sep=r'\\t', header=True)\n",
    "diagmd2022_df = spark.read.csv(path=constants.DIAGMD2022_FILEPATH, sep=r'\\t', header=True)\n",
    "procmd2022_df = spark.read.csv(path=constants.PROCMD2022_FILEPATH, sep=r'\\t', header=True)\n",
    "rx2022_df     = spark.read.csv(path=constants.RX2022_FILEPATH, sep=r'\\t', header=True)\n",
    "servbh2022_df = spark.read.csv(path=constants.SERVBH2022_FILEPATH, sep=r'\\t', header=True)\n",
    "servmd2022_df = spark.read.csv(path=constants.SERVMD2022_FILEPATH, sep=r'\\t', header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "65b083ff",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+---------+\n",
      "|  memid|dimMonthID|featureid|\n",
      "+-------+----------+---------+\n",
      "|1794474|  20210301|      d19|\n",
      "|  88972|  20210401|      d19|\n",
      "|1654929|  20211201|      d19|\n",
      "| 414065|  20211001|      d19|\n",
      "|1384890|  20210101|      d19|\n",
      "| 983331|  20210301|      d19|\n",
      "|1654929|  20210501|      d19|\n",
      "|1823456|  20210501|       d6|\n",
      "|1560573|  20210801|      d19|\n",
      "|1559339|  20211001|      d19|\n",
      "| 219885|  20211101|      d19|\n",
      "|1558160|  20210601|      d19|\n",
      "| 253024|  20210301|      d19|\n",
      "|1099536|  20210401|      d19|\n",
      "|1572285|  20210301|      d15|\n",
      "|1573212|  20210901|      d19|\n",
      "|1370945|  20211101|      d19|\n",
      "|1523991|  20210601|      d19|\n",
      "|1816312|  20211001|      d19|\n",
      "|1779469|  20211001|      d19|\n",
      "+-------+----------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "diagbh2021_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "a18c89f3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+----------+---------+\n",
      "|memid|dimMonthID|featureid|\n",
      "+-----+----------+---------+\n",
      "|   46|  20210101|      r79|\n",
      "|   46|  20210301|      r33|\n",
      "|   46|  20210301|      r37|\n",
      "|   46|  20210301|      r39|\n",
      "|   46|  20210301|      r49|\n",
      "|   46|  20210401|      r38|\n",
      "|   46|  20210401|      r39|\n",
      "|   46|  20210401|      r49|\n",
      "|   46|  20210401|      r79|\n",
      "|   46|  20210501|      r22|\n",
      "|   46|  20210501|      r32|\n",
      "|   46|  20210501|      r37|\n",
      "|   46|  20210501|      r49|\n",
      "|   46|  20210501|      r79|\n",
      "|   46|  20210601|      r27|\n",
      "|   46|  20210601|      r37|\n",
      "|   46|  20210601|      r41|\n",
      "|   46|  20210601|      r43|\n",
      "|   46|  20210701|      r37|\n",
      "|   46|  20210701|      r39|\n",
      "+-----+----------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "rx2021_df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b8f7774d",
   "metadata": {},
   "source": [
    "### Union consumption data into one dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "43d9b8ae",
   "metadata": {},
   "outputs": [],
   "source": [
    "all_data_dfs_list = [diagbh2021_df, diagmd2021_df, procmd2021_df, rx2021_df, servbh2021_df, servmd2021_df, diagbh2022_df, diagmd2022_df, procmd2022_df, rx2022_df, servbh2022_df, servmd2022_df]\n",
    "all_data_dfs = reduce(DataFrame.unionAll, all_data_dfs_list)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d2415bb5",
   "metadata": {},
   "source": [
    "### Select unique 'memid' , 'dimMonthID' pairs and create rank column "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "15188ff8",
   "metadata": {},
   "outputs": [],
   "source": [
    "memid_mths = all_data_dfs.select('memid', 'dimMonthID').drop_duplicates()\n",
    "windowSpec  = Window.partitionBy('memid').orderBy(\"dimMonthID\")\n",
    "memid_mths = memid_mths.withColumn('rnk', dense_rank().over(windowSpec))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "106db819",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 16:>                                                         (0 + 4) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+---+\n",
      "|  memid|dimMonthID|rnk|\n",
      "+-------+----------+---+\n",
      "|1002449|  20210101|  1|\n",
      "|1002449|  20210201|  2|\n",
      "|1002449|  20210301|  3|\n",
      "|1002449|  20210401|  4|\n",
      "|1002449|  20210501|  5|\n",
      "|1002449|  20210601|  6|\n",
      "|1002449|  20210701|  7|\n",
      "|1002449|  20210801|  8|\n",
      "|1002449|  20210901|  9|\n",
      "|1002449|  20211001| 10|\n",
      "|1002449|  20211101| 11|\n",
      "|1002449|  20211201| 12|\n",
      "|1002449|  20220101| 13|\n",
      "|1002449|  20220201| 14|\n",
      "|1002449|  20220301| 15|\n",
      "|1002449|  20220401| 16|\n",
      "|1002449|  20220501| 17|\n",
      "|1002449|  20220701| 18|\n",
      "|1002449|  20220801| 19|\n",
      "|1002449|  20220901| 20|\n",
      "+-------+----------+---+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "memid_mths.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4dcc362c",
   "metadata": {},
   "source": [
    "### Read featureids in different source files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "1f2683cd",
   "metadata": {},
   "outputs": [],
   "source": [
    "dummy_ids_diag = spark.read.csv(path=constants.DUMMY_IDS_DIAG_FILEPATH, sep=r'\\t', header=True)\n",
    "dummy_ids_proc = spark.read.csv(path=constants.DUMMY_IDS_PROC_FILEPATH, sep=r'\\t', header=True)\n",
    "dummy_ids_rx = spark.read.csv(path=constants.DUMMY_IDS_RX_FILEPATH, sep=r'\\t', header=True)\n",
    "dummy_ids_serv = spark.read.csv(path=constants.DUMMY_IDS_SERV_FILEPATH, sep=r'\\t', header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "05caf0ae",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+---------+-----------+--------------------+\n",
      "|level1dummyid|featureid|featuretype|              Level1|\n",
      "+-------------+---------+-----------+--------------------+\n",
      "|            1|       d1|       diag|Certain condition...|\n",
      "|            2|       d2|       diag|Certain infectiou...|\n",
      "|            3|       d3|       diag|Codes for special...|\n",
      "|            4|       d4|       diag|Congenital malfor...|\n",
      "|            5|       d5|       diag|Diseases of the b...|\n",
      "|            6|       d6|       diag|Diseases of the c...|\n",
      "|            7|       d7|       diag|Diseases of the d...|\n",
      "|            8|       d8|       diag|Diseases of the e...|\n",
      "|            9|       d9|       diag|Diseases of the e...|\n",
      "|           10|      d10|       diag|Diseases of the g...|\n",
      "|           11|      d11|       diag|Diseases of the m...|\n",
      "|           12|      d12|       diag|Diseases of the n...|\n",
      "|           13|      d13|       diag|Diseases of the r...|\n",
      "|           14|      d14|       diag|Diseases of the s...|\n",
      "|           15|      d15|       diag|Endocrine, nutrit...|\n",
      "|           16|      d16|       diag|External causes o...|\n",
      "|           17|      d17|       diag|Factors influenci...|\n",
      "|           18|      d18|       diag|Injury, poisoning...|\n",
      "|           19|      d19|       diag|Mental, Behaviora...|\n",
      "|           20|      d20|       diag| Neoplasms (C00-D49)|\n",
      "+-------------+---------+-----------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dummy_ids_diag.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "21e91ea8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+---------+-----------+--------------------+\n",
      "|level1dummyid|featureid|featuretype|              Level1|\n",
      "+-------------+---------+-----------+--------------------+\n",
      "|           11|      r11|         rx|         ANTIFUNGALS|\n",
      "|           81|      r81|         rx|DIETARY PRODUCTS/...|\n",
      "|           88|      r88|         rx|MOUTH/THROAT/DENT...|\n",
      "|           68|      r68|         rx|         GOUT AGENTS|\n",
      "|           99|      r99|         rx|    ASSORTED CLASSES|\n",
      "|           14|      r14|         rx|          AMEBICIDES|\n",
      "|           74|      r74|         rx|NEUROMUSCULAR AGENTS|\n",
      "|           61|      r61|         rx|ADHD/ANTI-NARCOLE...|\n",
      "|           40|      r40|         rx|CARDIOVASCULAR AG...|\n",
      "|            7|      r07|         rx|     AMINOGLYCOSIDES|\n",
      "|           35|      r35|         rx|     ANTIARRHYTHMICS|\n",
      "|           50|      r50|         rx|         ANTIEMETICS|\n",
      "|           85|      r85|         rx|HEMATOLOGICAL AGE...|\n",
      "|           64|      r64|         rx|ANALGESICS - NONN...|\n",
      "|           20|      r20|         rx|    BIOLOGICALS MISC|\n",
      "|           73|      r73|         rx|ANTIPARKINSON AGENTS|\n",
      "|           19|      r19|         rx|PASSIVE IMMUNIZIN...|\n",
      "|           16|      r16|         rx|ANTI-INFECTIVE AG...|\n",
      "|           95|      r95|         rx|ALTERNATIVE MEDIC...|\n",
      "|           83|      r83|         rx|      ANTICOAGULANTS|\n",
      "+-------------+---------+-----------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dummy_ids_rx.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a6b5ec70",
   "metadata": {},
   "source": [
    "### Union featureids into one dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "48b46633",
   "metadata": {},
   "outputs": [],
   "source": [
    "all_dummy_ids_list = [dummy_ids_diag, dummy_ids_proc, dummy_ids_rx, dummy_ids_serv]\n",
    "dummy_ids_df = reduce(DataFrame.unionAll, all_dummy_ids_list)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "74f2eb3b",
   "metadata": {},
   "source": [
    "### Order featureids by 'featuretype' and 'level1dummyid'. Cast a column for this to happen."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "9327bff9",
   "metadata": {},
   "outputs": [],
   "source": [
    "dummy_ids_df = dummy_ids_df.withColumn('level1dummyid', dummy_ids_df['level1dummyid'].cast(IntegerType()))\n",
    "dummy_ids_df = dummy_ids_df.orderBy('featuretype', 'level1dummyid')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "086f4f30",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+---------+-----------+--------------------+\n",
      "|level1dummyid|featureid|featuretype|              Level1|\n",
      "+-------------+---------+-----------+--------------------+\n",
      "|            1|       d1|       diag|Certain condition...|\n",
      "|            2|       d2|       diag|Certain infectiou...|\n",
      "|            3|       d3|       diag|Codes for special...|\n",
      "|            4|       d4|       diag|Congenital malfor...|\n",
      "|            5|       d5|       diag|Diseases of the b...|\n",
      "|            6|       d6|       diag|Diseases of the c...|\n",
      "|            7|       d7|       diag|Diseases of the d...|\n",
      "|            8|       d8|       diag|Diseases of the e...|\n",
      "|            9|       d9|       diag|Diseases of the e...|\n",
      "|           10|      d10|       diag|Diseases of the g...|\n",
      "|           11|      d11|       diag|Diseases of the m...|\n",
      "|           12|      d12|       diag|Diseases of the n...|\n",
      "|           13|      d13|       diag|Diseases of the r...|\n",
      "|           14|      d14|       diag|Diseases of the s...|\n",
      "|           15|      d15|       diag|Endocrine, nutrit...|\n",
      "|           16|      d16|       diag|External causes o...|\n",
      "|           17|      d17|       diag|Factors influenci...|\n",
      "|           18|      d18|       diag|Injury, poisoning...|\n",
      "|           19|      d19|       diag|Mental, Behaviora...|\n",
      "|           20|      d20|       diag| Neoplasms (C00-D49)|\n",
      "+-------------+---------+-----------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dummy_ids_df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0b816c8a",
   "metadata": {},
   "source": [
    "### Collect ordered features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "eb50329c",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "order_of_features = dummy_ids_df.rdd.map(lambda x : x.featureid).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "834e3fab",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['d1',\n",
       " 'd2',\n",
       " 'd3',\n",
       " 'd4',\n",
       " 'd5',\n",
       " 'd6',\n",
       " 'd7',\n",
       " 'd8',\n",
       " 'd9',\n",
       " 'd10',\n",
       " 'd11',\n",
       " 'd12',\n",
       " 'd13',\n",
       " 'd14',\n",
       " 'd15',\n",
       " 'd16',\n",
       " 'd17',\n",
       " 'd18',\n",
       " 'd19',\n",
       " 'd20',\n",
       " 'd21',\n",
       " 'd22',\n",
       " 'p1',\n",
       " 'p2',\n",
       " 'p3',\n",
       " 'p4',\n",
       " 'p5',\n",
       " 'p6',\n",
       " 'p7',\n",
       " 'p8',\n",
       " 'p9',\n",
       " 'p10',\n",
       " 'p11',\n",
       " 'p12',\n",
       " 'p13',\n",
       " 'p14',\n",
       " 'p15',\n",
       " 'p16',\n",
       " 'p17',\n",
       " 'r00',\n",
       " 'r01',\n",
       " 'r02',\n",
       " 'r03',\n",
       " 'r04',\n",
       " 'r05',\n",
       " 'r07',\n",
       " 'r08',\n",
       " 'r09',\n",
       " 'r11',\n",
       " 'r12',\n",
       " 'r13',\n",
       " 'r14',\n",
       " 'r15',\n",
       " 'r16',\n",
       " 'r17',\n",
       " 'r18',\n",
       " 'r19',\n",
       " 'r20',\n",
       " 'r21',\n",
       " 'r22',\n",
       " 'r23',\n",
       " 'r24',\n",
       " 'r25',\n",
       " 'r26',\n",
       " 'r27',\n",
       " 'r28',\n",
       " 'r29',\n",
       " 'r30',\n",
       " 'r31',\n",
       " 'r32',\n",
       " 'r33',\n",
       " 'r34',\n",
       " 'r35',\n",
       " 'r36',\n",
       " 'r37',\n",
       " 'r38',\n",
       " 'r39',\n",
       " 'r40',\n",
       " 'r41',\n",
       " 'r42',\n",
       " 'r43',\n",
       " 'r44',\n",
       " 'r45',\n",
       " 'r46',\n",
       " 'r47',\n",
       " 'r48',\n",
       " 'r49',\n",
       " 'r50',\n",
       " 'r51',\n",
       " 'r52',\n",
       " 'r53',\n",
       " 'r54',\n",
       " 'r55',\n",
       " 'r56',\n",
       " 'r57',\n",
       " 'r58',\n",
       " 'r59',\n",
       " 'r60',\n",
       " 'r61',\n",
       " 'r62',\n",
       " 'r64',\n",
       " 'r65',\n",
       " 'r66',\n",
       " 'r67',\n",
       " 'r68',\n",
       " 'r69',\n",
       " 'r70',\n",
       " 'r72',\n",
       " 'r73',\n",
       " 'r74',\n",
       " 'r75',\n",
       " 'r76',\n",
       " 'r77',\n",
       " 'r78',\n",
       " 'r79',\n",
       " 'r80',\n",
       " 'r81',\n",
       " 'r82',\n",
       " 'r83',\n",
       " 'r84',\n",
       " 'r85',\n",
       " 'r86',\n",
       " 'r87',\n",
       " 'r88',\n",
       " 'r89',\n",
       " 'r90',\n",
       " 'r92',\n",
       " 'r93',\n",
       " 'r94',\n",
       " 'r95',\n",
       " 'r96',\n",
       " 'r97',\n",
       " 'r98',\n",
       " 'r99',\n",
       " 's1',\n",
       " 's2',\n",
       " 's3',\n",
       " 's4',\n",
       " 's5',\n",
       " 's6',\n",
       " 's7',\n",
       " 's8',\n",
       " 's9',\n",
       " 's10',\n",
       " 's11',\n",
       " 's12',\n",
       " 's13',\n",
       " 's14',\n",
       " 's15',\n",
       " 's16',\n",
       " 's17',\n",
       " 's18',\n",
       " 's19',\n",
       " 's20',\n",
       " 's21',\n",
       " 's22',\n",
       " 's23',\n",
       " 's24',\n",
       " 's25',\n",
       " 's26',\n",
       " 's27',\n",
       " 's28',\n",
       " 's29',\n",
       " 's30',\n",
       " 's31',\n",
       " 's32',\n",
       " 's33',\n",
       " 's34',\n",
       " 's35',\n",
       " 's36',\n",
       " 's37',\n",
       " 's38',\n",
       " 's39']"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "order_of_features"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eaba19b3",
   "metadata": {},
   "source": [
    "### Crossjoin the unique 'memid', 'dimMonthID' pairs with featureids"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "0050540f",
   "metadata": {},
   "outputs": [],
   "source": [
    "memidmths_cross_dummy_ids = memid_mths.crossJoin(dummy_ids_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "7c16f7d2",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 32:================================================>       (19 + 3) / 22]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+---+-------------+---------+-----------+--------------------+\n",
      "|  memid|dimMonthID|rnk|level1dummyid|featureid|featuretype|              Level1|\n",
      "+-------+----------+---+-------------+---------+-----------+--------------------+\n",
      "|1002449|  20210101|  1|            1|       d1|       diag|Certain condition...|\n",
      "|1002449|  20210101|  1|            2|       d2|       diag|Certain infectiou...|\n",
      "|1002449|  20210101|  1|            3|       d3|       diag|Codes for special...|\n",
      "|1002449|  20210101|  1|            4|       d4|       diag|Congenital malfor...|\n",
      "|1002449|  20210101|  1|            5|       d5|       diag|Diseases of the b...|\n",
      "|1002449|  20210101|  1|            6|       d6|       diag|Diseases of the c...|\n",
      "|1002449|  20210101|  1|            7|       d7|       diag|Diseases of the d...|\n",
      "|1002449|  20210101|  1|            8|       d8|       diag|Diseases of the e...|\n",
      "|1002449|  20210101|  1|            9|       d9|       diag|Diseases of the e...|\n",
      "|1002449|  20210101|  1|           10|      d10|       diag|Diseases of the g...|\n",
      "|1002449|  20210101|  1|           11|      d11|       diag|Diseases of the m...|\n",
      "|1002449|  20210101|  1|           12|      d12|       diag|Diseases of the n...|\n",
      "|1002449|  20210101|  1|           13|      d13|       diag|Diseases of the r...|\n",
      "|1002449|  20210101|  1|           14|      d14|       diag|Diseases of the s...|\n",
      "|1002449|  20210101|  1|           15|      d15|       diag|Endocrine, nutrit...|\n",
      "|1002449|  20210101|  1|           16|      d16|       diag|External causes o...|\n",
      "|1002449|  20210101|  1|           17|      d17|       diag|Factors influenci...|\n",
      "|1002449|  20210101|  1|           18|      d18|       diag|Injury, poisoning...|\n",
      "|1002449|  20210101|  1|           19|      d19|       diag|Mental, Behaviora...|\n",
      "|1002449|  20210101|  1|           20|      d20|       diag| Neoplasms (C00-D49)|\n",
      "+-------+----------+---+-------------+---------+-----------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "memidmths_cross_dummy_ids.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "343eb692",
   "metadata": {},
   "source": [
    "### Rename columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "19f91eb6",
   "metadata": {},
   "outputs": [],
   "source": [
    "all_data_dfs = all_data_dfs.withColumnRenamed('memid', 'a_memid').withColumnRenamed('dimMonthID', 'a_dimMonthID').withColumnRenamed('featureid', 'a_featureid')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "49b82db3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- a_memid: string (nullable = true)\n",
      " |-- a_dimMonthID: string (nullable = true)\n",
      " |-- a_featureid: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "all_data_dfs.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "837b7e12",
   "metadata": {},
   "source": [
    "### Right join to create sparse consumption data flags for each member, for respective month, for each feature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "bd6c2100",
   "metadata": {},
   "outputs": [],
   "source": [
    "fj = memidmths_cross_dummy_ids.join(all_data_dfs, (memidmths_cross_dummy_ids['memid']==all_data_dfs['a_memid']) & (memidmths_cross_dummy_ids['dimMonthID']==all_data_dfs['a_dimMonthID']) & (memidmths_cross_dummy_ids['featureid']==all_data_dfs['a_featureid']), 'left')\n",
    "fjmod = fj.withColumn('is_feature_present', when(fj['a_memid'].isNull(), lit(0)).otherwise(lit(1))).drop(col('a_memid'), col('a_dimMonthID'), col('a_featureid'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "c5ead4f6",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+---+-------------+---------+-----------+--------------------+------------------+\n",
      "|  memid|dimMonthID|rnk|level1dummyid|featureid|featuretype|              Level1|is_feature_present|\n",
      "+-------+----------+---+-------------+---------+-----------+--------------------+------------------+\n",
      "|1000231|  20210101|  1|           11|      d11|       diag|Diseases of the m...|                 0|\n",
      "|1000231|  20210101|  1|           13|      d13|       diag|Diseases of the r...|                 1|\n",
      "|1000231|  20210101|  1|            5|       d5|       diag|Diseases of the b...|                 0|\n",
      "|1002449|  20210101|  1|           11|      d11|       diag|Diseases of the m...|                 0|\n",
      "|1002449|  20210101|  1|           14|      d14|       diag|Diseases of the s...|                 0|\n",
      "|1002449|  20210101|  1|           15|      d15|       diag|Endocrine, nutrit...|                 1|\n",
      "|1002449|  20210101|  1|           18|      d18|       diag|Injury, poisoning...|                 0|\n",
      "|1002449|  20210101|  1|           20|      d20|       diag| Neoplasms (C00-D49)|                 1|\n",
      "|1002449|  20210101|  1|            6|       d6|       diag|Diseases of the c...|                 1|\n",
      "|1000231|  20210101|  1|           15|      d15|       diag|Endocrine, nutrit...|                 1|\n",
      "|1000231|  20210101|  1|           16|      d16|       diag|External causes o...|                 0|\n",
      "|1000231|  20210101|  1|           17|      d17|       diag|Factors influenci...|                 1|\n",
      "|1000231|  20210101|  1|           19|      d19|       diag|Mental, Behaviora...|                 0|\n",
      "|1000231|  20210101|  1|            2|       d2|       diag|Certain infectiou...|                 0|\n",
      "|1000231|  20210101|  1|            4|       d4|       diag|Congenital malfor...|                 0|\n",
      "|1000231|  20210101|  1|            6|       d6|       diag|Diseases of the c...|                 1|\n",
      "|1000231|  20210101|  1|            7|       d7|       diag|Diseases of the d...|                 0|\n",
      "|1000231|  20210101|  1|            8|       d8|       diag|Diseases of the e...|                 0|\n",
      "|1002449|  20210101|  1|           10|      d10|       diag|Diseases of the g...|                 0|\n",
      "|1002449|  20210101|  1|           12|      d12|       diag|Diseases of the n...|                 0|\n",
      "+-------+----------+---+-------------+---------+-----------+--------------------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 55:>                                                         (0 + 2) / 2]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "fjmod.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a95bec1a",
   "metadata": {},
   "source": [
    "### For each member for respective month, create a list of consumption data flags"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "51447767",
   "metadata": {},
   "outputs": [],
   "source": [
    "windowSpec2 = Window.partitionBy('memid', 'dimMonthID').orderBy('featuretype', 'level1dummyid').rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)\n",
    "fjmod_short = fjmod.withColumn('feature_presence', collect_list('is_feature_present').over(windowSpec2))\n",
    "fjmod_short = fjmod_short.select('memid', 'dimMonthID', 'feature_presence').distinct()\n",
    "fjmod_short = fjmod_short.withColumn('feature_presence', fjmod_short['feature_presence'].cast(StringType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "c96c3fa9",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 69:>                                                         (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+--------------------+\n",
      "|  memid|dimMonthID|    feature_presence|\n",
      "+-------+----------+--------------------+\n",
      "|1000231|  20210501|[0, 0, 0, 0, 0, 1...|\n",
      "|1000428|  20210701|[0, 0, 0, 0, 0, 0...|\n",
      "|1000428|  20211201|[0, 1, 0, 0, 0, 0...|\n",
      "|1000442|  20210701|[0, 0, 0, 0, 0, 1...|\n",
      "|1002449|  20210101|[0, 0, 0, 0, 0, 1...|\n",
      "|1002449|  20210501|[0, 0, 0, 0, 0, 0...|\n",
      "|1002449|  20220901|[0, 0, 0, 0, 0, 1...|\n",
      "|1002449|  20221201|[0, 0, 0, 0, 0, 1...|\n",
      "|1002593|  20210101|[0, 0, 0, 0, 0, 1...|\n",
      "|1002593|  20211101|[0, 0, 0, 0, 0, 1...|\n",
      "|1002655|  20210201|[0, 0, 0, 0, 0, 0...|\n",
      "|1002655|  20210601|[0, 0, 0, 0, 0, 0...|\n",
      "|1002655|  20211001|[0, 0, 0, 0, 0, 0...|\n",
      "|1002655|  20211101|[0, 0, 0, 0, 0, 0...|\n",
      "|1002655|  20220101|[0, 0, 0, 0, 0, 0...|\n",
      "|1002655|  20220901|[0, 0, 0, 0, 0, 0...|\n",
      "|1002655|  20221101|[0, 0, 0, 0, 0, 0...|\n",
      "|1002695|  20220301|[0, 0, 0, 0, 0, 0...|\n",
      "|1003468|  20220201|[0, 0, 0, 0, 1, 0...|\n",
      "| 100354|  20210501|[0, 0, 0, 0, 1, 1...|\n",
      "+-------+----------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "fjmod_short.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8cbf88ca",
   "metadata": {},
   "source": [
    "### Partition by 'memid' and write data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "852c845d",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "fjmod_short.write.option('header',True).partitionBy('memid').mode(\"overwrite\").csv(\"consumption_data\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cd34c8b4",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}