PySpark Dataframe Complete Guide (with COVID-19 Dataset).ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark Dataframe Complete Guide (with COVID-19 Dataset)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark which is one of the most used tools when it comes to working with Big Data."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "While once upon a time Spark used to be heavily reliant on RDD manipulations, Spark has now provided a DataFrame API for us Data Scientists to work with. [Doc](https://docs.databricks.com/spark/latest/dataframes-datasets/introduction-to-dataframes-python.html#).\n",
    "\n",
    "**Yay~!**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this notebook, We will learn standard Spark functionalities needed to work with DataFrames, and finally some tips to handle the inevitable errors you will face."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "I'm going to skip the Spark Installation part for the sake of the notebook, so please go to [Apache Spark Website](http://spark.apache.org/downloads.html) to install Spark that are right to your work setting."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "#we use the findspark library to locate spark on our local machine\n",
    "import findspark\n",
    "findspark.init('C:/Users/bokhy/spark/spark-2.4.6-bin-hadoop2.7')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "from datetime import date, timedelta, datetime\n",
    "import time\n",
    "\n",
    "import pyspark # only run this after findspark.init()\n",
    "from pyspark.sql import SparkSession, SQLContext\n",
    "from pyspark.context import SparkContext\n",
    "from pyspark.sql.functions import * \n",
    "from pyspark.sql.types import * "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Initiate the Spark Session\n",
    "spark = SparkSession.builder.appName('covid-example').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://DESKTOP-L9PS4IG:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.6</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>covid-example</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x1e615704fc8>"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data\n",
    " We will be working with the Data Science for COVID-19 in South Korea, which is one of the most detailed datasets on the internet for COVID."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Data can be found in this kaggle URL [Link](https://www.kaggle.com/kimjihoo/coronavirusdataset)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1. Basic Functions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [1] Load (Read) the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "cases = spark.read.load(\"./data/Case.csv\",\n",
    "                        format=\"csv\", \n",
    "                        sep=\",\", \n",
    "                        inferSchema=\"true\", \n",
    "                        header=\"true\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------+---------------+-----+--------------------+---------+---------+----------+\n",
      "| case_id|province|           city|group|      infection_case|confirmed| latitude| longitude|\n",
      "+--------+--------+---------------+-----+--------------------+---------+---------+----------+\n",
      "| 1000001|   Seoul|     Yongsan-gu| true|       Itaewon Clubs|      139|37.538621|126.992652|\n",
      "| 1000002|   Seoul|      Gwanak-gu| true|             Richway|      119| 37.48208|126.901384|\n",
      "| 1000003|   Seoul|        Guro-gu| true| Guro-gu Call Center|       95|37.508163|126.884387|\n",
      "| 1000004|   Seoul|   Yangcheon-gu| true|Yangcheon Table T...|       43|37.546061|126.874209|\n",
      "| 1000005|   Seoul|      Dobong-gu| true|     Day Care Center|       43|37.679422|127.044374|\n",
      "| 1000006|   Seoul|        Guro-gu| true|Manmin Central Ch...|       41|37.481059|126.894343|\n",
      "| 1000007|   Seoul|from other city| true|SMR Newly Planted...|       36|        -|         -|\n",
      "| 1000008|   Seoul|  Dongdaemun-gu| true|       Dongan Church|       17|37.592888|127.056766|\n",
      "| 1000009|   Seoul|from other city| true|Coupang Logistics...|       25|        -|         -|\n",
      "| 1000010|   Seoul|      Gwanak-gu| true|     Wangsung Church|       30|37.481735|126.930121|\n",
      "| 1000011|   Seoul|   Eunpyeong-gu| true|Eunpyeong St. Mar...|       14| 37.63369|  126.9165|\n",
      "| 1000012|   Seoul|   Seongdong-gu| true|    Seongdong-gu APT|       13| 37.55713|  127.0403|\n",
      "| 1000013|   Seoul|      Jongno-gu| true|Jongno Community ...|       10| 37.57681|   127.006|\n",
      "| 1000014|   Seoul|     Gangnam-gu| true|Samsung Medical C...|        7| 37.48825| 127.08559|\n",
      "| 1000015|   Seoul|        Jung-gu| true|Jung-gu Fashion C...|        7|37.562405|126.984377|\n",
      "| 1000016|   Seoul|   Seodaemun-gu| true|  Yeonana News Class|        5|37.558147|126.943799|\n",
      "| 1000017|   Seoul|      Jongno-gu| true|Korea Campus Crus...|        7|37.594782|126.968022|\n",
      "| 1000018|   Seoul|     Gangnam-gu| true|Gangnam Yeoksam-d...|        6|        -|         -|\n",
      "| 1000019|   Seoul|from other city| true|Daejeon door-to-d...|        1|        -|         -|\n",
      "| 1000020|   Seoul|   Geumcheon-gu| true|Geumcheon-gu rice...|        6|        -|         -|\n",
      "+--------+--------+---------------+-----+--------------------+---------+---------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# First few rows in the file\n",
    "cases.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It looks ok right now, but sometimes as we the number of columns increases, the formatting becomes not too great. I have noticed that the following trick helps in displaying in pandas format in my Jupyter Notebook. \n",
    "\n",
    "The **.toPandas()** function converts a **Spark Dataframe** into a **Pandas Dataframe**, which is much easier to play with."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>case_id</th>\n",
       "      <th>province</th>\n",
       "      <th>city</th>\n",
       "      <th>group</th>\n",
       "      <th>infection_case</th>\n",
       "      <th>confirmed</th>\n",
       "      <th>latitude</th>\n",
       "      <th>longitude</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1000001</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Yongsan-gu</td>\n",
       "      <td>True</td>\n",
       "      <td>Itaewon Clubs</td>\n",
       "      <td>139</td>\n",
       "      <td>37.538621</td>\n",
       "      <td>126.992652</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1000002</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Gwanak-gu</td>\n",
       "      <td>True</td>\n",
       "      <td>Richway</td>\n",
       "      <td>119</td>\n",
       "      <td>37.48208</td>\n",
       "      <td>126.901384</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1000003</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Guro-gu</td>\n",
       "      <td>True</td>\n",
       "      <td>Guro-gu Call Center</td>\n",
       "      <td>95</td>\n",
       "      <td>37.508163</td>\n",
       "      <td>126.884387</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1000004</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Yangcheon-gu</td>\n",
       "      <td>True</td>\n",
       "      <td>Yangcheon Table Tennis Club</td>\n",
       "      <td>43</td>\n",
       "      <td>37.546061</td>\n",
       "      <td>126.874209</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>1000005</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Dobong-gu</td>\n",
       "      <td>True</td>\n",
       "      <td>Day Care Center</td>\n",
       "      <td>43</td>\n",
       "      <td>37.679422</td>\n",
       "      <td>127.044374</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>1000006</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Guro-gu</td>\n",
       "      <td>True</td>\n",
       "      <td>Manmin Central Church</td>\n",
       "      <td>41</td>\n",
       "      <td>37.481059</td>\n",
       "      <td>126.894343</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>1000007</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>from other city</td>\n",
       "      <td>True</td>\n",
       "      <td>SMR Newly Planted Churches Group</td>\n",
       "      <td>36</td>\n",
       "      <td>-</td>\n",
       "      <td>-</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>1000008</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Dongdaemun-gu</td>\n",
       "      <td>True</td>\n",
       "      <td>Dongan Church</td>\n",
       "      <td>17</td>\n",
       "      <td>37.592888</td>\n",
       "      <td>127.056766</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>1000009</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>from other city</td>\n",
       "      <td>True</td>\n",
       "      <td>Coupang Logistics Center</td>\n",
       "      <td>25</td>\n",
       "      <td>-</td>\n",
       "      <td>-</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>1000010</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Gwanak-gu</td>\n",
       "      <td>True</td>\n",
       "      <td>Wangsung Church</td>\n",
       "      <td>30</td>\n",
       "      <td>37.481735</td>\n",
       "      <td>126.930121</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "    case_id province             city  group  \\\n",
       "0   1000001    Seoul       Yongsan-gu   True   \n",
       "1   1000002    Seoul        Gwanak-gu   True   \n",
       "2   1000003    Seoul          Guro-gu   True   \n",
       "3   1000004    Seoul     Yangcheon-gu   True   \n",
       "4   1000005    Seoul        Dobong-gu   True   \n",
       "5   1000006    Seoul          Guro-gu   True   \n",
       "6   1000007    Seoul  from other city   True   \n",
       "7   1000008    Seoul    Dongdaemun-gu   True   \n",
       "8   1000009    Seoul  from other city   True   \n",
       "9   1000010    Seoul        Gwanak-gu   True   \n",
       "\n",
       "                     infection_case  confirmed   latitude   longitude  \n",
       "0                     Itaewon Clubs        139  37.538621  126.992652  \n",
       "1                           Richway        119   37.48208  126.901384  \n",
       "2               Guro-gu Call Center         95  37.508163  126.884387  \n",
       "3       Yangcheon Table Tennis Club         43  37.546061  126.874209  \n",
       "4                   Day Care Center         43  37.679422  127.044374  \n",
       "5             Manmin Central Church         41  37.481059  126.894343  \n",
       "6  SMR Newly Planted Churches Group         36          -           -  \n",
       "7                     Dongan Church         17  37.592888  127.056766  \n",
       "8          Coupang Logistics Center         25          -           -  \n",
       "9                   Wangsung Church         30  37.481735  126.930121  "
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cases.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [2] Change Column Names"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To change a single column,"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "cases = cases.withColumnRenamed(\"infection_case\",\"infection_source\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To change all columns,"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "cases = cases.toDF(*['case_id', 'province', 'city', 'group', 'infection_case', 'confirmed',\n",
    "       'latitude', 'longitude'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------+---------------+-----+--------------------+---------+---------+----------+\n",
      "|case_id|province|           city|group|      infection_case|confirmed| latitude| longitude|\n",
      "+-------+--------+---------------+-----+--------------------+---------+---------+----------+\n",
      "|1000001|   Seoul|     Yongsan-gu| true|       Itaewon Clubs|      139|37.538621|126.992652|\n",
      "|1000002|   Seoul|      Gwanak-gu| true|             Richway|      119| 37.48208|126.901384|\n",
      "|1000003|   Seoul|        Guro-gu| true| Guro-gu Call Center|       95|37.508163|126.884387|\n",
      "|1000004|   Seoul|   Yangcheon-gu| true|Yangcheon Table T...|       43|37.546061|126.874209|\n",
      "|1000005|   Seoul|      Dobong-gu| true|     Day Care Center|       43|37.679422|127.044374|\n",
      "|1000006|   Seoul|        Guro-gu| true|Manmin Central Ch...|       41|37.481059|126.894343|\n",
      "|1000007|   Seoul|from other city| true|SMR Newly Planted...|       36|        -|         -|\n",
      "|1000008|   Seoul|  Dongdaemun-gu| true|       Dongan Church|       17|37.592888|127.056766|\n",
      "|1000009|   Seoul|from other city| true|Coupang Logistics...|       25|        -|         -|\n",
      "|1000010|   Seoul|      Gwanak-gu| true|     Wangsung Church|       30|37.481735|126.930121|\n",
      "|1000011|   Seoul|   Eunpyeong-gu| true|Eunpyeong St. Mar...|       14| 37.63369|  126.9165|\n",
      "|1000012|   Seoul|   Seongdong-gu| true|    Seongdong-gu APT|       13| 37.55713|  127.0403|\n",
      "|1000013|   Seoul|      Jongno-gu| true|Jongno Community ...|       10| 37.57681|   127.006|\n",
      "|1000014|   Seoul|     Gangnam-gu| true|Samsung Medical C...|        7| 37.48825| 127.08559|\n",
      "|1000015|   Seoul|        Jung-gu| true|Jung-gu Fashion C...|        7|37.562405|126.984377|\n",
      "|1000016|   Seoul|   Seodaemun-gu| true|  Yeonana News Class|        5|37.558147|126.943799|\n",
      "|1000017|   Seoul|      Jongno-gu| true|Korea Campus Crus...|        7|37.594782|126.968022|\n",
      "|1000018|   Seoul|     Gangnam-gu| true|Gangnam Yeoksam-d...|        6|        -|         -|\n",
      "|1000019|   Seoul|from other city| true|Daejeon door-to-d...|        1|        -|         -|\n",
      "|1000020|   Seoul|   Geumcheon-gu| true|Geumcheon-gu rice...|        6|        -|         -|\n",
      "+-------+--------+---------------+-----+--------------------+---------+---------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "cases.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [3] Change Column Names"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can select a subset of columns using the **select** "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------------+--------------------+---------+\n",
      "|province|           city|      infection_case|confirmed|\n",
      "+--------+---------------+--------------------+---------+\n",
      "|   Seoul|     Yongsan-gu|       Itaewon Clubs|      139|\n",
      "|   Seoul|      Gwanak-gu|             Richway|      119|\n",
      "|   Seoul|        Guro-gu| Guro-gu Call Center|       95|\n",
      "|   Seoul|   Yangcheon-gu|Yangcheon Table T...|       43|\n",
      "|   Seoul|      Dobong-gu|     Day Care Center|       43|\n",
      "|   Seoul|        Guro-gu|Manmin Central Ch...|       41|\n",
      "|   Seoul|from other city|SMR Newly Planted...|       36|\n",
      "|   Seoul|  Dongdaemun-gu|       Dongan Church|       17|\n",
      "|   Seoul|from other city|Coupang Logistics...|       25|\n",
      "|   Seoul|      Gwanak-gu|     Wangsung Church|       30|\n",
      "|   Seoul|   Eunpyeong-gu|Eunpyeong St. Mar...|       14|\n",
      "|   Seoul|   Seongdong-gu|    Seongdong-gu APT|       13|\n",
      "|   Seoul|      Jongno-gu|Jongno Community ...|       10|\n",
      "|   Seoul|     Gangnam-gu|Samsung Medical C...|        7|\n",
      "|   Seoul|        Jung-gu|Jung-gu Fashion C...|        7|\n",
      "|   Seoul|   Seodaemun-gu|  Yeonana News Class|        5|\n",
      "|   Seoul|      Jongno-gu|Korea Campus Crus...|        7|\n",
      "|   Seoul|     Gangnam-gu|Gangnam Yeoksam-d...|        6|\n",
      "|   Seoul|from other city|Daejeon door-to-d...|        1|\n",
      "|   Seoul|   Geumcheon-gu|Geumcheon-gu rice...|        6|\n",
      "+--------+---------------+--------------------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "cases = cases.select('province','city','infection_case','confirmed')\n",
    "cases.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [4] Sort by Column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+---------------+--------------------+---------+\n",
      "|         province|           city|      infection_case|confirmed|\n",
      "+-----------------+---------------+--------------------+---------+\n",
      "|          Jeju-do|              -|contact with patient|        0|\n",
      "|       Gangwon-do|              -|contact with patient|        0|\n",
      "|            Seoul|     Gangseo-gu|SJ Investment Cal...|        0|\n",
      "|            Busan|from other city|Cheongdo Daenam H...|        1|\n",
      "|     Jeollabuk-do|from other city|  Shincheonji Church|        1|\n",
      "|            Seoul|from other city|Anyang Gunpo Past...|        1|\n",
      "|            Seoul|     Gangnam-gu|Gangnam Dongin Ch...|        1|\n",
      "|           Sejong|from other city|  Shincheonji Church|        1|\n",
      "|     Jeollanam-do|from other city|  Shincheonji Church|        1|\n",
      "|          Jeju-do|from other city|       Itaewon Clubs|        1|\n",
      "|            Seoul|from other city|Daejeon door-to-d...|        1|\n",
      "|            Seoul|              -|         Orange Life|        1|\n",
      "|           Sejong|              -|                 etc|        1|\n",
      "|          Gwangju|              -|                 etc|        1|\n",
      "|            Daegu|from other city|       Itaewon Clubs|        2|\n",
      "|            Daegu|from other city|Cheongdo Daenam H...|        2|\n",
      "|          Incheon|from other city|  Shincheonji Church|        2|\n",
      "|            Seoul|from other city|Uiwang Logistics ...|        2|\n",
      "|          Daejeon|from other city|  Shincheonji Church|        2|\n",
      "|Chungcheongbuk-do|from other city| Guro-gu Call Center|        2|\n",
      "+-----------------+---------------+--------------------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Simple sort\n",
    "cases.sort(\"confirmed\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+---------------+--------------------+---------+\n",
      "|         province|           city|      infection_case|confirmed|\n",
      "+-----------------+---------------+--------------------+---------+\n",
      "|            Daegu|         Nam-gu|  Shincheonji Church|     4511|\n",
      "|            Daegu|              -|contact with patient|      917|\n",
      "|            Daegu|              -|                 etc|      747|\n",
      "| Gyeongsangbuk-do|from other city|  Shincheonji Church|      566|\n",
      "|      Gyeonggi-do|              -|     overseas inflow|      305|\n",
      "|            Seoul|              -|     overseas inflow|      298|\n",
      "|            Daegu|   Dalseong-gun|Second Mi-Ju Hosp...|      196|\n",
      "| Gyeongsangbuk-do|              -|contact with patient|      190|\n",
      "|            Seoul|              -|contact with patient|      162|\n",
      "|            Seoul|     Yongsan-gu|       Itaewon Clubs|      139|\n",
      "| Gyeongsangbuk-do|              -|                 etc|      133|\n",
      "|            Daegu|         Seo-gu|Hansarang Convale...|      124|\n",
      "|            Seoul|      Gwanak-gu|             Richway|      119|\n",
      "| Gyeongsangbuk-do|   Cheongdo-gun|Cheongdo Daenam H...|      119|\n",
      "|Chungcheongnam-do|     Cheonan-si|gym facility in C...|      103|\n",
      "|            Daegu|   Dalseong-gun|Daesil Convalesce...|      101|\n",
      "|            Seoul|              -|                 etc|      100|\n",
      "|            Seoul|        Guro-gu| Guro-gu Call Center|       95|\n",
      "|      Gyeonggi-do|              -|                 etc|       84|\n",
      "|          Incheon|              -|     overseas inflow|       68|\n",
      "+-----------------+---------------+--------------------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Descending Sort\n",
    "from pyspark.sql import functions as F\n",
    "\n",
    "cases.sort(F.desc(\"confirmed\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [5] Change Column Type"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------------+--------------------+---------+\n",
      "|province|           city|      infection_case|confirmed|\n",
      "+--------+---------------+--------------------+---------+\n",
      "|   Seoul|     Yongsan-gu|       Itaewon Clubs|      139|\n",
      "|   Seoul|      Gwanak-gu|             Richway|      119|\n",
      "|   Seoul|        Guro-gu| Guro-gu Call Center|       95|\n",
      "|   Seoul|   Yangcheon-gu|Yangcheon Table T...|       43|\n",
      "|   Seoul|      Dobong-gu|     Day Care Center|       43|\n",
      "|   Seoul|        Guro-gu|Manmin Central Ch...|       41|\n",
      "|   Seoul|from other city|SMR Newly Planted...|       36|\n",
      "|   Seoul|  Dongdaemun-gu|       Dongan Church|       17|\n",
      "|   Seoul|from other city|Coupang Logistics...|       25|\n",
      "|   Seoul|      Gwanak-gu|     Wangsung Church|       30|\n",
      "|   Seoul|   Eunpyeong-gu|Eunpyeong St. Mar...|       14|\n",
      "|   Seoul|   Seongdong-gu|    Seongdong-gu APT|       13|\n",
      "|   Seoul|      Jongno-gu|Jongno Community ...|       10|\n",
      "|   Seoul|     Gangnam-gu|Samsung Medical C...|        7|\n",
      "|   Seoul|        Jung-gu|Jung-gu Fashion C...|        7|\n",
      "|   Seoul|   Seodaemun-gu|  Yeonana News Class|        5|\n",
      "|   Seoul|      Jongno-gu|Korea Campus Crus...|        7|\n",
      "|   Seoul|     Gangnam-gu|Gangnam Yeoksam-d...|        6|\n",
      "|   Seoul|from other city|Daejeon door-to-d...|        1|\n",
      "|   Seoul|   Geumcheon-gu|Geumcheon-gu rice...|        6|\n",
      "+--------+---------------+--------------------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.types import DoubleType, IntegerType, StringType\n",
    "\n",
    "cases = cases.withColumn('confirmed', F.col('confirmed').cast(IntegerType()))\n",
    "cases = cases.withColumn('city', F.col('city').cast(StringType()))\n",
    "\n",
    "cases.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [6] Filter "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can filter a data frame using multiple conditions using AND(&), OR(|) and NOT(~) conditions. For example, we may want to find out all the different infection_case in Daegu with more than 10 confirmed cases."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------------+--------------------+---------+\n",
      "|province|        city|      infection_case|confirmed|\n",
      "+--------+------------+--------------------+---------+\n",
      "|   Daegu|      Nam-gu|  Shincheonji Church|     4511|\n",
      "|   Daegu|Dalseong-gun|Second Mi-Ju Hosp...|      196|\n",
      "|   Daegu|      Seo-gu|Hansarang Convale...|      124|\n",
      "|   Daegu|Dalseong-gun|Daesil Convalesce...|      101|\n",
      "|   Daegu|     Dong-gu|     Fatima Hospital|       39|\n",
      "|   Daegu|           -|     overseas inflow|       41|\n",
      "|   Daegu|           -|contact with patient|      917|\n",
      "|   Daegu|           -|                 etc|      747|\n",
      "+--------+------------+--------------------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "cases.filter((cases.confirmed>10) & (cases.province=='Daegu')).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [7] GroupBy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+---------------+--------------+--------------+\n",
      "|        province|           city|sum(confirmed)|max(confirmed)|\n",
      "+----------------+---------------+--------------+--------------+\n",
      "|Gyeongsangnam-do|       Jinju-si|             9|             9|\n",
      "|           Seoul|        Guro-gu|           139|            95|\n",
      "|           Seoul|     Gangnam-gu|            18|             7|\n",
      "|         Daejeon|              -|           100|            55|\n",
      "|    Jeollabuk-do|from other city|             6|             3|\n",
      "|Gyeongsangnam-do|Changnyeong-gun|             7|             7|\n",
      "|           Seoul|              -|           561|           298|\n",
      "|         Jeju-do|from other city|             1|             1|\n",
      "|Gyeongsangbuk-do|              -|           345|           190|\n",
      "|Gyeongsangnam-do|   Geochang-gun|            18|            10|\n",
      "|Gyeongsangbuk-do|        Gumi-si|            10|            10|\n",
      "|         Incheon|from other city|           117|            53|\n",
      "|           Busan|              -|            85|            36|\n",
      "|           Daegu|         Seo-gu|           124|           124|\n",
      "|           Busan|     Suyeong-gu|             5|             5|\n",
      "|     Gyeonggi-do|   Uijeongbu-si|            50|            50|\n",
      "|           Seoul|     Yongsan-gu|           139|           139|\n",
      "|           Daegu|              -|          1705|           917|\n",
      "|           Seoul|   Seodaemun-gu|             5|             5|\n",
      "|     Gyeonggi-do|    Seongnam-si|            94|            67|\n",
      "+----------------+---------------+--------------+--------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import functions as F\n",
    "\n",
    "cases.groupBy([\"province\",\"city\"]).agg(F.sum(\"confirmed\") ,F.max(\"confirmed\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Or if we don’t like the new column names, we can use the **alias** keyword to rename columns in the agg command itself."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+---------------+--------------+-----------------------+\n",
      "|        province|           city|TotalConfirmed|MaxFromOneConfirmedCase|\n",
      "+----------------+---------------+--------------+-----------------------+\n",
      "|Gyeongsangnam-do|       Jinju-si|             9|                      9|\n",
      "|           Seoul|        Guro-gu|           139|                     95|\n",
      "|           Seoul|     Gangnam-gu|            18|                      7|\n",
      "|         Daejeon|              -|           100|                     55|\n",
      "|    Jeollabuk-do|from other city|             6|                      3|\n",
      "|Gyeongsangnam-do|Changnyeong-gun|             7|                      7|\n",
      "|           Seoul|              -|           561|                    298|\n",
      "|         Jeju-do|from other city|             1|                      1|\n",
      "|Gyeongsangbuk-do|              -|           345|                    190|\n",
      "|Gyeongsangnam-do|   Geochang-gun|            18|                     10|\n",
      "|Gyeongsangbuk-do|        Gumi-si|            10|                     10|\n",
      "|         Incheon|from other city|           117|                     53|\n",
      "|           Busan|              -|            85|                     36|\n",
      "|           Daegu|         Seo-gu|           124|                    124|\n",
      "|           Busan|     Suyeong-gu|             5|                      5|\n",
      "|     Gyeonggi-do|   Uijeongbu-si|            50|                     50|\n",
      "|           Seoul|     Yongsan-gu|           139|                    139|\n",
      "|           Daegu|              -|          1705|                    917|\n",
      "|           Seoul|   Seodaemun-gu|             5|                      5|\n",
      "|     Gyeonggi-do|    Seongnam-si|            94|                     67|\n",
      "+----------------+---------------+--------------+-----------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "cases.groupBy([\"province\",\"city\"]).agg(\n",
    "    F.sum(\"confirmed\").alias(\"TotalConfirmed\"),\\\n",
    "    F.max(\"confirmed\").alias(\"MaxFromOneConfirmedCase\")\\\n",
    "    ).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [8] Joins"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here, We will go with the region file which contains region information such as elementary_school_count, elderly_population_ratio, etc."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>code</th>\n",
       "      <th>province</th>\n",
       "      <th>city</th>\n",
       "      <th>latitude</th>\n",
       "      <th>longitude</th>\n",
       "      <th>elementary_school_count</th>\n",
       "      <th>kindergarten_count</th>\n",
       "      <th>university_count</th>\n",
       "      <th>academy_ratio</th>\n",
       "      <th>elderly_population_ratio</th>\n",
       "      <th>elderly_alone_ratio</th>\n",
       "      <th>nursing_home_count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>10000</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>37.566953</td>\n",
       "      <td>126.977977</td>\n",
       "      <td>607</td>\n",
       "      <td>830</td>\n",
       "      <td>48</td>\n",
       "      <td>1.44</td>\n",
       "      <td>15.38</td>\n",
       "      <td>5.8</td>\n",
       "      <td>22739</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>10010</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Gangnam-gu</td>\n",
       "      <td>37.518421</td>\n",
       "      <td>127.047222</td>\n",
       "      <td>33</td>\n",
       "      <td>38</td>\n",
       "      <td>0</td>\n",
       "      <td>4.18</td>\n",
       "      <td>13.17</td>\n",
       "      <td>4.3</td>\n",
       "      <td>3088</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>10020</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Gangdong-gu</td>\n",
       "      <td>37.530492</td>\n",
       "      <td>127.123837</td>\n",
       "      <td>27</td>\n",
       "      <td>32</td>\n",
       "      <td>0</td>\n",
       "      <td>1.54</td>\n",
       "      <td>14.55</td>\n",
       "      <td>5.4</td>\n",
       "      <td>1023</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>10030</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Gangbuk-gu</td>\n",
       "      <td>37.639938</td>\n",
       "      <td>127.025508</td>\n",
       "      <td>14</td>\n",
       "      <td>21</td>\n",
       "      <td>0</td>\n",
       "      <td>0.67</td>\n",
       "      <td>19.49</td>\n",
       "      <td>8.5</td>\n",
       "      <td>628</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>10040</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Gangseo-gu</td>\n",
       "      <td>37.551166</td>\n",
       "      <td>126.849506</td>\n",
       "      <td>36</td>\n",
       "      <td>56</td>\n",
       "      <td>1</td>\n",
       "      <td>1.17</td>\n",
       "      <td>14.39</td>\n",
       "      <td>5.7</td>\n",
       "      <td>1080</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>10050</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Gwanak-gu</td>\n",
       "      <td>37.478290</td>\n",
       "      <td>126.951502</td>\n",
       "      <td>22</td>\n",
       "      <td>33</td>\n",
       "      <td>1</td>\n",
       "      <td>0.89</td>\n",
       "      <td>15.12</td>\n",
       "      <td>4.9</td>\n",
       "      <td>909</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>10060</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Gwangjin-gu</td>\n",
       "      <td>37.538712</td>\n",
       "      <td>127.082366</td>\n",
       "      <td>22</td>\n",
       "      <td>33</td>\n",
       "      <td>3</td>\n",
       "      <td>1.16</td>\n",
       "      <td>13.75</td>\n",
       "      <td>4.8</td>\n",
       "      <td>723</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>10070</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Guro-gu</td>\n",
       "      <td>37.495632</td>\n",
       "      <td>126.887650</td>\n",
       "      <td>26</td>\n",
       "      <td>34</td>\n",
       "      <td>3</td>\n",
       "      <td>1.00</td>\n",
       "      <td>16.21</td>\n",
       "      <td>5.7</td>\n",
       "      <td>741</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>10080</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Geumcheon-gu</td>\n",
       "      <td>37.456852</td>\n",
       "      <td>126.895229</td>\n",
       "      <td>18</td>\n",
       "      <td>19</td>\n",
       "      <td>0</td>\n",
       "      <td>0.96</td>\n",
       "      <td>16.15</td>\n",
       "      <td>6.7</td>\n",
       "      <td>475</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>10090</td>\n",
       "      <td>Seoul</td>\n",
       "      <td>Nowon-gu</td>\n",
       "      <td>37.654259</td>\n",
       "      <td>127.056294</td>\n",
       "      <td>42</td>\n",
       "      <td>66</td>\n",
       "      <td>6</td>\n",
       "      <td>1.39</td>\n",
       "      <td>15.40</td>\n",
       "      <td>7.4</td>\n",
       "      <td>952</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "    code province          city   latitude   longitude  \\\n",
       "0  10000    Seoul         Seoul  37.566953  126.977977   \n",
       "1  10010    Seoul    Gangnam-gu  37.518421  127.047222   \n",
       "2  10020    Seoul   Gangdong-gu  37.530492  127.123837   \n",
       "3  10030    Seoul    Gangbuk-gu  37.639938  127.025508   \n",
       "4  10040    Seoul    Gangseo-gu  37.551166  126.849506   \n",
       "5  10050    Seoul     Gwanak-gu  37.478290  126.951502   \n",
       "6  10060    Seoul   Gwangjin-gu  37.538712  127.082366   \n",
       "7  10070    Seoul       Guro-gu  37.495632  126.887650   \n",
       "8  10080    Seoul  Geumcheon-gu  37.456852  126.895229   \n",
       "9  10090    Seoul      Nowon-gu  37.654259  127.056294   \n",
       "\n",
       "   elementary_school_count  kindergarten_count  university_count  \\\n",
       "0                      607                 830                48   \n",
       "1                       33                  38                 0   \n",
       "2                       27                  32                 0   \n",
       "3                       14                  21                 0   \n",
       "4                       36                  56                 1   \n",
       "5                       22                  33                 1   \n",
       "6                       22                  33                 3   \n",
       "7                       26                  34                 3   \n",
       "8                       18                  19                 0   \n",
       "9                       42                  66                 6   \n",
       "\n",
       "   academy_ratio  elderly_population_ratio  elderly_alone_ratio  \\\n",
       "0           1.44                     15.38                  5.8   \n",
       "1           4.18                     13.17                  4.3   \n",
       "2           1.54                     14.55                  5.4   \n",
       "3           0.67                     19.49                  8.5   \n",
       "4           1.17                     14.39                  5.7   \n",
       "5           0.89                     15.12                  4.9   \n",
       "6           1.16                     13.75                  4.8   \n",
       "7           1.00                     16.21                  5.7   \n",
       "8           0.96                     16.15                  6.7   \n",
       "9           1.39                     15.40                  7.4   \n",
       "\n",
       "   nursing_home_count  \n",
       "0               22739  \n",
       "1                3088  \n",
       "2                1023  \n",
       "3                 628  \n",
       "4                1080  \n",
       "5                 909  \n",
       "6                 723  \n",
       "7                 741  \n",
       "8                 475  \n",
       "9                 952  "
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "regions = spark.read.load(\"./data/Region.csv\",\n",
    "                          format=\"csv\", \n",
    "                          sep=\",\", \n",
    "                          inferSchema=\"true\", \n",
    "                          header=\"true\")\n",
    "\n",
    "regions.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>province</th>\n",
       "      <th>city</th>\n",
       "      <th>infection_case</th>\n",
       "      <th>confirmed</th>\n",
       "      <th>code</th>\n",
       "      <th>latitude</th>\n",
       "      <th>longitude</th>\n",
       "      <th>elementary_school_count</th>\n",
       "      <th>kindergarten_count</th>\n",
       "      <th>university_count</th>\n",
       "      <th>academy_ratio</th>\n",
       "      <th>elderly_population_ratio</th>\n",
       "      <th>elderly_alone_ratio</th>\n",
       "      <th>nursing_home_count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Seoul</td>\n",
       "      <td>Yongsan-gu</td>\n",
       "      <td>Itaewon Clubs</td>\n",
       "      <td>139</td>\n",
       "      <td>10210.0</td>\n",
       "      <td>37.532768</td>\n",
       "      <td>126.990021</td>\n",
       "      <td>15.0</td>\n",
       "      <td>13.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.68</td>\n",
       "      <td>16.87</td>\n",
       "      <td>6.5</td>\n",
       "      <td>435.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Seoul</td>\n",
       "      <td>Gwanak-gu</td>\n",
       "      <td>Richway</td>\n",
       "      <td>119</td>\n",
       "      <td>10050.0</td>\n",
       "      <td>37.478290</td>\n",
       "      <td>126.951502</td>\n",
       "      <td>22.0</td>\n",
       "      <td>33.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.89</td>\n",
       "      <td>15.12</td>\n",
       "      <td>4.9</td>\n",
       "      <td>909.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Seoul</td>\n",
       "      <td>Guro-gu</td>\n",
       "      <td>Guro-gu Call Center</td>\n",
       "      <td>95</td>\n",
       "      <td>10070.0</td>\n",
       "      <td>37.495632</td>\n",
       "      <td>126.887650</td>\n",
       "      <td>26.0</td>\n",
       "      <td>34.0</td>\n",
       "      <td>3.0</td>\n",
       "      <td>1.00</td>\n",
       "      <td>16.21</td>\n",
       "      <td>5.7</td>\n",
       "      <td>741.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>Seoul</td>\n",
       "      <td>Yangcheon-gu</td>\n",
       "      <td>Yangcheon Table Tennis Club</td>\n",
       "      <td>43</td>\n",
       "      <td>10190.0</td>\n",
       "      <td>37.517189</td>\n",
       "      <td>126.866618</td>\n",
       "      <td>30.0</td>\n",
       "      <td>43.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>2.26</td>\n",
       "      <td>13.55</td>\n",
       "      <td>5.5</td>\n",
       "      <td>816.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>Seoul</td>\n",
       "      <td>Dobong-gu</td>\n",
       "      <td>Day Care Center</td>\n",
       "      <td>43</td>\n",
       "      <td>10100.0</td>\n",
       "      <td>37.668952</td>\n",
       "      <td>127.047082</td>\n",
       "      <td>23.0</td>\n",
       "      <td>26.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.95</td>\n",
       "      <td>17.89</td>\n",
       "      <td>7.2</td>\n",
       "      <td>485.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>Seoul</td>\n",
       "      <td>Guro-gu</td>\n",
       "      <td>Manmin Central Church</td>\n",
       "      <td>41</td>\n",
       "      <td>10070.0</td>\n",
       "      <td>37.495632</td>\n",
       "      <td>126.887650</td>\n",
       "      <td>26.0</td>\n",
       "      <td>34.0</td>\n",
       "      <td>3.0</td>\n",
       "      <td>1.00</td>\n",
       "      <td>16.21</td>\n",
       "      <td>5.7</td>\n",
       "      <td>741.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>Seoul</td>\n",
       "      <td>from other city</td>\n",
       "      <td>SMR Newly Planted Churches Group</td>\n",
       "      <td>36</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>Seoul</td>\n",
       "      <td>Dongdaemun-gu</td>\n",
       "      <td>Dongan Church</td>\n",
       "      <td>17</td>\n",
       "      <td>10110.0</td>\n",
       "      <td>37.574552</td>\n",
       "      <td>127.039721</td>\n",
       "      <td>21.0</td>\n",
       "      <td>31.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>1.06</td>\n",
       "      <td>17.26</td>\n",
       "      <td>6.7</td>\n",
       "      <td>832.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>Seoul</td>\n",
       "      <td>from other city</td>\n",
       "      <td>Coupang Logistics Center</td>\n",
       "      <td>25</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>Seoul</td>\n",
       "      <td>Gwanak-gu</td>\n",
       "      <td>Wangsung Church</td>\n",
       "      <td>30</td>\n",
       "      <td>10050.0</td>\n",
       "      <td>37.478290</td>\n",
       "      <td>126.951502</td>\n",
       "      <td>22.0</td>\n",
       "      <td>33.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.89</td>\n",
       "      <td>15.12</td>\n",
       "      <td>4.9</td>\n",
       "      <td>909.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  province             city                    infection_case  confirmed  \\\n",
       "0    Seoul       Yongsan-gu                     Itaewon Clubs        139   \n",
       "1    Seoul        Gwanak-gu                           Richway        119   \n",
       "2    Seoul          Guro-gu               Guro-gu Call Center         95   \n",
       "3    Seoul     Yangcheon-gu       Yangcheon Table Tennis Club         43   \n",
       "4    Seoul        Dobong-gu                   Day Care Center         43   \n",
       "5    Seoul          Guro-gu             Manmin Central Church         41   \n",
       "6    Seoul  from other city  SMR Newly Planted Churches Group         36   \n",
       "7    Seoul    Dongdaemun-gu                     Dongan Church         17   \n",
       "8    Seoul  from other city          Coupang Logistics Center         25   \n",
       "9    Seoul        Gwanak-gu                   Wangsung Church         30   \n",
       "\n",
       "      code   latitude   longitude  elementary_school_count  \\\n",
       "0  10210.0  37.532768  126.990021                     15.0   \n",
       "1  10050.0  37.478290  126.951502                     22.0   \n",
       "2  10070.0  37.495632  126.887650                     26.0   \n",
       "3  10190.0  37.517189  126.866618                     30.0   \n",
       "4  10100.0  37.668952  127.047082                     23.0   \n",
       "5  10070.0  37.495632  126.887650                     26.0   \n",
       "6      NaN        NaN         NaN                      NaN   \n",
       "7  10110.0  37.574552  127.039721                     21.0   \n",
       "8      NaN        NaN         NaN                      NaN   \n",
       "9  10050.0  37.478290  126.951502                     22.0   \n",
       "\n",
       "   kindergarten_count  university_count  academy_ratio  \\\n",
       "0                13.0               1.0           0.68   \n",
       "1                33.0               1.0           0.89   \n",
       "2                34.0               3.0           1.00   \n",
       "3                43.0               0.0           2.26   \n",
       "4                26.0               1.0           0.95   \n",
       "5                34.0               3.0           1.00   \n",
       "6                 NaN               NaN            NaN   \n",
       "7                31.0               4.0           1.06   \n",
       "8                 NaN               NaN            NaN   \n",
       "9                33.0               1.0           0.89   \n",
       "\n",
       "   elderly_population_ratio  elderly_alone_ratio  nursing_home_count  \n",
       "0                     16.87                  6.5               435.0  \n",
       "1                     15.12                  4.9               909.0  \n",
       "2                     16.21                  5.7               741.0  \n",
       "3                     13.55                  5.5               816.0  \n",
       "4                     17.89                  7.2               485.0  \n",
       "5                     16.21                  5.7               741.0  \n",
       "6                       NaN                  NaN                 NaN  \n",
       "7                     17.26                  6.7               832.0  \n",
       "8                       NaN                  NaN                 NaN  \n",
       "9                     15.12                  4.9               909.0  "
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Left Join 'Case' with 'Region' on Province and City column\n",
    "cases = cases.join(regions, ['province','city'],how='left')\n",
    "cases.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. Use SQL with DataFrames"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We first register the cases dataframe to a temporary table cases_table on which we can run SQL operations. As you can see, the result of the SQL select statement is again a Spark Dataframe.\n",
    "\n",
    "All complex SQL queries like GROUP BY, HAVING, AND ORDER BY clauses can be applied in 'Sql' function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cases.registerTempTable('cases_table')\n",
    "newDF = SQLContext.sql('select * from cases_table where confirmed > 100')\n",
    "newDF.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3. Create New Columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There are many ways that you can use to create a column in a PySpark Dataframe."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [1] Using Spark Native Functions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can use .withcolumn along with PySpark SQL functions to create a new column. In essence, you can find String functions, Date functions, and Math functions already implemented using Spark functions. Our first function, the F.col function gives us access to the column. So if we wanted to add 100 to a column, we could use F.col as:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------------+--------------------+---------+-----+---------+----------+-----------------------+------------------+----------------+-------------+------------------------+-------------------+------------------+------------+\n",
      "|province|           city|      infection_case|confirmed| code| latitude| longitude|elementary_school_count|kindergarten_count|university_count|academy_ratio|elderly_population_ratio|elderly_alone_ratio|nursing_home_count|NewConfirmed|\n",
      "+--------+---------------+--------------------+---------+-----+---------+----------+-----------------------+------------------+----------------+-------------+------------------------+-------------------+------------------+------------+\n",
      "|   Seoul|     Yongsan-gu|       Itaewon Clubs|      139|10210|37.532768|126.990021|                     15|                13|               1|         0.68|                   16.87|                6.5|               435|         239|\n",
      "|   Seoul|      Gwanak-gu|             Richway|      119|10050| 37.47829|126.951502|                     22|                33|               1|         0.89|                   15.12|                4.9|               909|         219|\n",
      "|   Seoul|        Guro-gu| Guro-gu Call Center|       95|10070|37.495632| 126.88765|                     26|                34|               3|          1.0|                   16.21|                5.7|               741|         195|\n",
      "|   Seoul|   Yangcheon-gu|Yangcheon Table T...|       43|10190|37.517189|126.866618|                     30|                43|               0|         2.26|                   13.55|                5.5|               816|         143|\n",
      "|   Seoul|      Dobong-gu|     Day Care Center|       43|10100|37.668952|127.047082|                     23|                26|               1|         0.95|                   17.89|                7.2|               485|         143|\n",
      "|   Seoul|        Guro-gu|Manmin Central Ch...|       41|10070|37.495632| 126.88765|                     26|                34|               3|          1.0|                   16.21|                5.7|               741|         141|\n",
      "|   Seoul|from other city|SMR Newly Planted...|       36| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|         136|\n",
      "|   Seoul|  Dongdaemun-gu|       Dongan Church|       17|10110|37.574552|127.039721|                     21|                31|               4|         1.06|                   17.26|                6.7|               832|         117|\n",
      "|   Seoul|from other city|Coupang Logistics...|       25| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|         125|\n",
      "|   Seoul|      Gwanak-gu|     Wangsung Church|       30|10050| 37.47829|126.951502|                     22|                33|               1|         0.89|                   15.12|                4.9|               909|         130|\n",
      "|   Seoul|   Eunpyeong-gu|Eunpyeong St. Mar...|       14|10220|37.603481|126.929173|                     31|                44|               1|         1.09|                    17.0|                6.5|               874|         114|\n",
      "|   Seoul|   Seongdong-gu|    Seongdong-gu APT|       13|10160|37.563277|127.036647|                     21|                30|               2|         0.97|                   14.76|                5.3|               593|         113|\n",
      "|   Seoul|      Jongno-gu|Jongno Community ...|       10|10230|37.572999|126.979189|                     13|                17|               3|         1.71|                   18.27|                6.8|               668|         110|\n",
      "|   Seoul|     Gangnam-gu|Samsung Medical C...|        7|10010|37.518421|127.047222|                     33|                38|               0|         4.18|                   13.17|                4.3|              3088|         107|\n",
      "|   Seoul|        Jung-gu|Jung-gu Fashion C...|        7|10240|37.563988| 126.99753|                     12|                14|               2|         0.94|                   18.42|                7.4|               728|         107|\n",
      "|   Seoul|   Seodaemun-gu|  Yeonana News Class|        5|10140|37.579428|126.936771|                     19|                25|               6|         1.12|                   16.77|                6.2|               587|         105|\n",
      "|   Seoul|      Jongno-gu|Korea Campus Crus...|        7|10230|37.572999|126.979189|                     13|                17|               3|         1.71|                   18.27|                6.8|               668|         107|\n",
      "|   Seoul|     Gangnam-gu|Gangnam Yeoksam-d...|        6|10010|37.518421|127.047222|                     33|                38|               0|         4.18|                   13.17|                4.3|              3088|         106|\n",
      "|   Seoul|from other city|Daejeon door-to-d...|        1| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|         101|\n",
      "|   Seoul|   Geumcheon-gu|Geumcheon-gu rice...|        6|10080|37.456852|126.895229|                     18|                19|               0|         0.96|                   16.15|                6.7|               475|         106|\n",
      "+--------+---------------+--------------------+---------+-----+---------+----------+-----------------------+------------------+----------------+-------------+------------------------+-------------------+------------------+------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import pyspark.sql.functions as F\n",
    "\n",
    "casesWithNewConfirmed = cases.withColumn(\"NewConfirmed\", 100 + F.col(\"confirmed\"))\n",
    "casesWithNewConfirmed.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also use math functions like F.exp function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------------+--------------------+---------+-----+---------+----------+-----------------------+------------------+----------------+-------------+------------------------+-------------------+------------------+--------------------+\n",
      "|province|           city|      infection_case|confirmed| code| latitude| longitude|elementary_school_count|kindergarten_count|university_count|academy_ratio|elderly_population_ratio|elderly_alone_ratio|nursing_home_count|        ExpConfirmed|\n",
      "+--------+---------------+--------------------+---------+-----+---------+----------+-----------------------+------------------+----------------+-------------+------------------------+-------------------+------------------+--------------------+\n",
      "|   Seoul|     Yongsan-gu|       Itaewon Clubs|      139|10210|37.532768|126.990021|                     15|                13|               1|         0.68|                   16.87|                6.5|               435|2.327732040478862...|\n",
      "|   Seoul|      Gwanak-gu|             Richway|      119|10050| 37.47829|126.951502|                     22|                33|               1|         0.89|                   15.12|                4.9|               909|4.797813327299302E51|\n",
      "|   Seoul|        Guro-gu| Guro-gu Call Center|       95|10070|37.495632| 126.88765|                     26|                34|               3|          1.0|                   16.21|                5.7|               741|1.811239082889023...|\n",
      "|   Seoul|   Yangcheon-gu|Yangcheon Table T...|       43|10190|37.517189|126.866618|                     30|                43|               0|         2.26|                   13.55|                5.5|               816|4.727839468229346...|\n",
      "|   Seoul|      Dobong-gu|     Day Care Center|       43|10100|37.668952|127.047082|                     23|                26|               1|         0.95|                   17.89|                7.2|               485|4.727839468229346...|\n",
      "|   Seoul|        Guro-gu|Manmin Central Ch...|       41|10070|37.495632| 126.88765|                     26|                34|               3|          1.0|                   16.21|                5.7|               741|6.398434935300549...|\n",
      "|   Seoul|from other city|SMR Newly Planted...|       36| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|4.311231547115195E15|\n",
      "|   Seoul|  Dongdaemun-gu|       Dongan Church|       17|10110|37.574552|127.039721|                     21|                31|               4|         1.06|                   17.26|                6.7|               832|  2.41549527535753E7|\n",
      "|   Seoul|from other city|Coupang Logistics...|       25| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|7.200489933738588E10|\n",
      "|   Seoul|      Gwanak-gu|     Wangsung Church|       30|10050| 37.47829|126.951502|                     22|                33|               1|         0.89|                   15.12|                4.9|               909|1.068647458152446...|\n",
      "|   Seoul|   Eunpyeong-gu|Eunpyeong St. Mar...|       14|10220|37.603481|126.929173|                     31|                44|               1|         1.09|                    17.0|                6.5|               874|  1202604.2841647768|\n",
      "|   Seoul|   Seongdong-gu|    Seongdong-gu APT|       13|10160|37.563277|127.036647|                     21|                30|               2|         0.97|                   14.76|                5.3|               593|   442413.3920089205|\n",
      "|   Seoul|      Jongno-gu|Jongno Community ...|       10|10230|37.572999|126.979189|                     13|                17|               3|         1.71|                   18.27|                6.8|               668|  22026.465794806718|\n",
      "|   Seoul|     Gangnam-gu|Samsung Medical C...|        7|10010|37.518421|127.047222|                     33|                38|               0|         4.18|                   13.17|                4.3|              3088|  1096.6331584284585|\n",
      "|   Seoul|        Jung-gu|Jung-gu Fashion C...|        7|10240|37.563988| 126.99753|                     12|                14|               2|         0.94|                   18.42|                7.4|               728|  1096.6331584284585|\n",
      "|   Seoul|   Seodaemun-gu|  Yeonana News Class|        5|10140|37.579428|126.936771|                     19|                25|               6|         1.12|                   16.77|                6.2|               587|   148.4131591025766|\n",
      "|   Seoul|      Jongno-gu|Korea Campus Crus...|        7|10230|37.572999|126.979189|                     13|                17|               3|         1.71|                   18.27|                6.8|               668|  1096.6331584284585|\n",
      "|   Seoul|     Gangnam-gu|Gangnam Yeoksam-d...|        6|10010|37.518421|127.047222|                     33|                38|               0|         4.18|                   13.17|                4.3|              3088|   403.4287934927351|\n",
      "|   Seoul|from other city|Daejeon door-to-d...|        1| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   2.718281828459045|\n",
      "|   Seoul|   Geumcheon-gu|Geumcheon-gu rice...|        6|10080|37.456852|126.895229|                     18|                19|               0|         0.96|                   16.15|                6.7|               475|   403.4287934927351|\n",
      "+--------+---------------+--------------------+---------+-----+---------+----------+-----------------------+------------------+----------------+-------------+------------------------+-------------------+------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "casesWithExpConfirmed = cases.withColumn(\"ExpConfirmed\", F.exp(\"confirmed\"))\n",
    "casesWithExpConfirmed.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [2] Using Spark UDFs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sometimes we want to do complicated things to a column or multiple columns. This could be thought of as a map operation on a PySpark Dataframe to a single column or multiple columns. While Spark SQL functions do solve many use cases when it comes to column creation, I use Spark UDF whenever I need more matured Python functionality. \\\n",
    "\n",
    "To use Spark UDFs, we need to use the F.udf function to convert a regular python function to a Spark UDF. We also need to specify the return type of the function. In this example the return type is StringType()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------------+--------------------+---------+-----+---------+----------+-----------------------+------------------+----------------+-------------+------------------------+-------------------+------------------+-------+\n",
      "|province|           city|      infection_case|confirmed| code| latitude| longitude|elementary_school_count|kindergarten_count|university_count|academy_ratio|elderly_population_ratio|elderly_alone_ratio|nursing_home_count|HighLow|\n",
      "+--------+---------------+--------------------+---------+-----+---------+----------+-----------------------+------------------+----------------+-------------+------------------------+-------------------+------------------+-------+\n",
      "|   Seoul|     Yongsan-gu|       Itaewon Clubs|      139|10210|37.532768|126.990021|                     15|                13|               1|         0.68|                   16.87|                6.5|               435|   high|\n",
      "|   Seoul|      Gwanak-gu|             Richway|      119|10050| 37.47829|126.951502|                     22|                33|               1|         0.89|                   15.12|                4.9|               909|   high|\n",
      "|   Seoul|        Guro-gu| Guro-gu Call Center|       95|10070|37.495632| 126.88765|                     26|                34|               3|          1.0|                   16.21|                5.7|               741|   high|\n",
      "|   Seoul|   Yangcheon-gu|Yangcheon Table T...|       43|10190|37.517189|126.866618|                     30|                43|               0|         2.26|                   13.55|                5.5|               816|    low|\n",
      "|   Seoul|      Dobong-gu|     Day Care Center|       43|10100|37.668952|127.047082|                     23|                26|               1|         0.95|                   17.89|                7.2|               485|    low|\n",
      "|   Seoul|        Guro-gu|Manmin Central Ch...|       41|10070|37.495632| 126.88765|                     26|                34|               3|          1.0|                   16.21|                5.7|               741|    low|\n",
      "|   Seoul|from other city|SMR Newly Planted...|       36| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|    low|\n",
      "|   Seoul|  Dongdaemun-gu|       Dongan Church|       17|10110|37.574552|127.039721|                     21|                31|               4|         1.06|                   17.26|                6.7|               832|    low|\n",
      "|   Seoul|from other city|Coupang Logistics...|       25| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|    low|\n",
      "|   Seoul|      Gwanak-gu|     Wangsung Church|       30|10050| 37.47829|126.951502|                     22|                33|               1|         0.89|                   15.12|                4.9|               909|    low|\n",
      "|   Seoul|   Eunpyeong-gu|Eunpyeong St. Mar...|       14|10220|37.603481|126.929173|                     31|                44|               1|         1.09|                    17.0|                6.5|               874|    low|\n",
      "|   Seoul|   Seongdong-gu|    Seongdong-gu APT|       13|10160|37.563277|127.036647|                     21|                30|               2|         0.97|                   14.76|                5.3|               593|    low|\n",
      "|   Seoul|      Jongno-gu|Jongno Community ...|       10|10230|37.572999|126.979189|                     13|                17|               3|         1.71|                   18.27|                6.8|               668|    low|\n",
      "|   Seoul|     Gangnam-gu|Samsung Medical C...|        7|10010|37.518421|127.047222|                     33|                38|               0|         4.18|                   13.17|                4.3|              3088|    low|\n",
      "|   Seoul|        Jung-gu|Jung-gu Fashion C...|        7|10240|37.563988| 126.99753|                     12|                14|               2|         0.94|                   18.42|                7.4|               728|    low|\n",
      "|   Seoul|   Seodaemun-gu|  Yeonana News Class|        5|10140|37.579428|126.936771|                     19|                25|               6|         1.12|                   16.77|                6.2|               587|    low|\n",
      "|   Seoul|      Jongno-gu|Korea Campus Crus...|        7|10230|37.572999|126.979189|                     13|                17|               3|         1.71|                   18.27|                6.8|               668|    low|\n",
      "|   Seoul|     Gangnam-gu|Gangnam Yeoksam-d...|        6|10010|37.518421|127.047222|                     33|                38|               0|         4.18|                   13.17|                4.3|              3088|    low|\n",
      "|   Seoul|from other city|Daejeon door-to-d...|        1| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|    low|\n",
      "|   Seoul|   Geumcheon-gu|Geumcheon-gu rice...|        6|10080|37.456852|126.895229|                     18|                19|               0|         0.96|                   16.15|                6.7|               475|    low|\n",
      "+--------+---------------+--------------------+---------+-----+---------+----------+-----------------------+------------------+----------------+-------------+------------------------+-------------------+------------------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.types import *\n",
    "\n",
    "def casesHighLow(confirmed):\n",
    "    if confirmed < 50: \n",
    "        return 'low'\n",
    "    else:\n",
    "        return 'high'\n",
    "    \n",
    "#convert to a UDF Function by passing in the function and return type of function\n",
    "casesHighLowUDF = F.udf(casesHighLow, StringType())\n",
    "CasesWithHighLow = cases.withColumn(\"HighLow\", casesHighLowUDF(\"confirmed\"))\n",
    "CasesWithHighLow.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [3] Using Pandas UDF"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This allows you to use pandas functionality with Spark. I generally use it when I have to run a groupBy operation on a Spark dataframe or whenever I need to create rolling features\n",
    " \n",
    "The way we use it is by using the F.pandas_udf decorator. **We assume here that the input to the function will be a pandas data frame**\n",
    "\n",
    "The only complexity here is that we have to provide a schema for the output Dataframe. We can use the original schema of a dataframe to create the outSchema."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- province: string (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- infection_case: string (nullable = true)\n",
      " |-- confirmed: integer (nullable = true)\n",
      " |-- code: integer (nullable = true)\n",
      " |-- latitude: double (nullable = true)\n",
      " |-- longitude: double (nullable = true)\n",
      " |-- elementary_school_count: integer (nullable = true)\n",
      " |-- kindergarten_count: integer (nullable = true)\n",
      " |-- university_count: integer (nullable = true)\n",
      " |-- academy_ratio: double (nullable = true)\n",
      " |-- elderly_population_ratio: double (nullable = true)\n",
      " |-- elderly_alone_ratio: double (nullable = true)\n",
      " |-- nursing_home_count: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "cases.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[case_id: int, province: string, city: string, group: boolean, infection_case: string, confirmed: int, latitude: string, longitude: string, normalized_confirmed: double]"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.types import IntegerType, StringType, DoubleType, BooleanType\n",
    "from pyspark.sql.types import StructType, StructField\n",
    "\n",
    "# Declare the schema for the output of our function\n",
    "\n",
    "outSchema = StructType([StructField('case_id',IntegerType(),True),\n",
    "                        StructField('province',StringType(),True),\n",
    "                        StructField('city',StringType(),True),\n",
    "                        StructField('group',BooleanType(),True),\n",
    "                        StructField('infection_case',StringType(),True),\n",
    "                        StructField('confirmed',IntegerType(),True),\n",
    "                        StructField('latitude',StringType(),True),\n",
    "                        StructField('longitude',StringType(),True),\n",
    "                        StructField('normalized_confirmed',DoubleType(),True)\n",
    "                       ])\n",
    "# decorate our function with pandas_udf decorator\n",
    "@F.pandas_udf(outSchema, F.PandasUDFType.GROUPED_MAP)\n",
    "def subtract_mean(pdf):\n",
    "    # pdf is a pandas.DataFrame\n",
    "    v = pdf.confirmed\n",
    "    v = v - v.mean()\n",
    "    pdf['normalized_confirmed'] = v\n",
    "    return pdf\n",
    "\n",
    "confirmed_groupwise_normalization = cases.groupby(\"infection_case\").apply(subtract_mean)\n",
    "\n",
    "confirmed_groupwise_normalization.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4. Spark Window Functions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will simply look at some of the most important and useful window functions available."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----+-----------------+---------+--------+--------+\n",
      "|               date|time|         province|confirmed|released|deceased|\n",
      "+-------------------+----+-----------------+---------+--------+--------+\n",
      "|2020-01-20 00:00:00|  16|            Seoul|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|            Busan|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|            Daegu|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|          Incheon|        1|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|          Gwangju|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|          Daejeon|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|            Ulsan|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|           Sejong|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|      Gyeonggi-do|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|       Gangwon-do|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|Chungcheongbuk-do|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|Chungcheongnam-do|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|     Jeollabuk-do|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|     Jeollanam-do|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16| Gyeongsangbuk-do|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16| Gyeongsangnam-do|        0|       0|       0|\n",
      "|2020-01-20 00:00:00|  16|          Jeju-do|        0|       0|       0|\n",
      "|2020-01-21 00:00:00|  16|            Seoul|        0|       0|       0|\n",
      "|2020-01-21 00:00:00|  16|            Busan|        0|       0|       0|\n",
      "|2020-01-21 00:00:00|  16|            Daegu|        0|       0|       0|\n",
      "+-------------------+----+-----------------+---------+--------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "timeprovince = spark.read.load(\"./data/TimeProvince.csv\",\n",
    "                          format=\"csv\", \n",
    "                          sep=\",\", \n",
    "                          inferSchema=\"true\", \n",
    "                          header=\"true\")\n",
    "\n",
    "timeprovince.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [1] Ranking"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can get rank as well as dense_rank on a group using this function. For example, you may want to have a column in your cases table that provides the rank of infection_case based on the number of infection_case in a province. We can do this by:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+---------------+--------------------+---------+-----+---------+----------+-----------------------+------------------+----------------+-------------+------------------------+-------------------+------------------+----+\n",
      "|         province|           city|      infection_case|confirmed| code| latitude| longitude|elementary_school_count|kindergarten_count|university_count|academy_ratio|elderly_population_ratio|elderly_alone_ratio|nursing_home_count|rank|\n",
      "+-----------------+---------------+--------------------+---------+-----+---------+----------+-----------------------+------------------+----------------+-------------+------------------------+-------------------+------------------+----+\n",
      "|           Sejong|         Sejong|Ministry of Ocean...|       31|17000|36.480132|127.289021|                     48|                60|               3|         1.78|                    9.48|                3.8|               491|   1|\n",
      "|           Sejong|         Sejong|gym facility in S...|        8|17000|36.480132|127.289021|                     48|                60|               3|         1.78|                    9.48|                3.8|               491|   2|\n",
      "|           Sejong|              -|     overseas inflow|        5| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   3|\n",
      "|           Sejong|              -|contact with patient|        3| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   4|\n",
      "|           Sejong|from other city|  Shincheonji Church|        1| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   5|\n",
      "|           Sejong|              -|                 etc|        1| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   5|\n",
      "|            Ulsan|              -|     overseas inflow|       25| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   1|\n",
      "|            Ulsan|from other city|  Shincheonji Church|       16| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   2|\n",
      "|            Ulsan|              -|                 etc|        7| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   3|\n",
      "|            Ulsan|              -|contact with patient|        3| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   4|\n",
      "|Chungcheongbuk-do|              -|     overseas inflow|       13| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   1|\n",
      "|Chungcheongbuk-do|     Goesan-gun|Goesan-gun Jangye...|       11|40010| 36.81534|127.786651|                     14|                15|               1|         0.36|                   33.01|               16.5|                64|   2|\n",
      "|Chungcheongbuk-do|              -|                 etc|       11| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   2|\n",
      "|Chungcheongbuk-do|from other city|       Itaewon Clubs|        9| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   4|\n",
      "|Chungcheongbuk-do|              -|contact with patient|        8| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   5|\n",
      "|Chungcheongbuk-do|from other city|  Shincheonji Church|        6| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   6|\n",
      "|Chungcheongbuk-do|from other city| Guro-gu Call Center|        2| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   7|\n",
      "|       Gangwon-do|from other city|  Shincheonji Church|       17| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   1|\n",
      "|       Gangwon-do|              -|     overseas inflow|       16| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   2|\n",
      "|       Gangwon-do|from other city|Uijeongbu St. Mar...|       10| null|     null|      null|                   null|              null|            null|         null|                    null|               null|              null|   3|\n",
      "+-----------------+---------------+--------------------+---------+-----+---------+----------+-----------------------+------------------+----------------+-------------+------------------------+-------------------+------------------+----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.window import Window\n",
    "windowSpec = Window().partitionBy(['province']).orderBy(F.desc('confirmed'))\n",
    "cases.withColumn(\"rank\",F.rank().over(windowSpec)).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [2] Lag Variables"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sometimes our data science models may need **lag based** features. For example, a model might have variables like the price last week or sales quantity the previous day. We can create such features using the lag function with window functions. \\\n",
    "\n",
    "Here I am trying to get the confirmed cases 7 days before. I am filtering to show the results as the first few days of corona cases were zeros. You can see here that the lag_7 day feature is shifted by 7 days."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----+--------+---------+--------+--------+-----+\n",
      "|               date|time|province|confirmed|released|deceased|lag_7|\n",
      "+-------------------+----+--------+---------+--------+--------+-----+\n",
      "|2020-03-10 00:00:00|   0|  Sejong|        8|       0|       0|    1|\n",
      "|2020-03-11 00:00:00|   0|  Sejong|       10|       0|       0|    1|\n",
      "|2020-03-12 00:00:00|   0|  Sejong|       15|       0|       0|    1|\n",
      "|2020-03-13 00:00:00|   0|  Sejong|       32|       0|       0|    1|\n",
      "|2020-03-14 00:00:00|   0|  Sejong|       38|       0|       0|    2|\n",
      "|2020-03-15 00:00:00|   0|  Sejong|       39|       0|       0|    3|\n",
      "|2020-03-16 00:00:00|   0|  Sejong|       40|       0|       0|    6|\n",
      "|2020-03-17 00:00:00|   0|  Sejong|       40|       0|       0|    8|\n",
      "|2020-03-18 00:00:00|   0|  Sejong|       41|       0|       0|   10|\n",
      "|2020-03-19 00:00:00|   0|  Sejong|       41|       0|       0|   15|\n",
      "|2020-03-20 00:00:00|   0|  Sejong|       41|       0|       0|   32|\n",
      "|2020-03-21 00:00:00|   0|  Sejong|       41|       2|       0|   38|\n",
      "|2020-03-22 00:00:00|   0|  Sejong|       41|       3|       0|   39|\n",
      "|2020-03-23 00:00:00|   0|  Sejong|       42|       3|       0|   40|\n",
      "|2020-03-24 00:00:00|   0|  Sejong|       42|       3|       0|   40|\n",
      "|2020-03-25 00:00:00|   0|  Sejong|       44|       3|       0|   41|\n",
      "|2020-03-26 00:00:00|   0|  Sejong|       44|       8|       0|   41|\n",
      "|2020-03-27 00:00:00|   0|  Sejong|       44|       9|       0|   41|\n",
      "|2020-03-28 00:00:00|   0|  Sejong|       44|       9|       0|   41|\n",
      "|2020-03-29 00:00:00|   0|  Sejong|       46|      11|       0|   41|\n",
      "+-------------------+----+--------+---------+--------+--------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.window import Window\n",
    "\n",
    "windowSpec = Window().partitionBy(['province']).orderBy('date')\n",
    "\n",
    "timeprovinceWithLag = timeprovince.withColumn(\"lag_7\",F.lag(\"confirmed\", 7).over(windowSpec))\n",
    "\n",
    "timeprovinceWithLag.filter(timeprovinceWithLag.date>'2020-03-10').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [3] Rolling Aggregations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For example, we might want to have a rolling 7-day sales sum/mean as a feature for our sales regression model. Let us calculate the rolling mean of confirmed cases for the last 7 days here. This is what a lot of the people are already doing with this dataset to see the real trends."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----+--------+---------+--------+--------+------------------+\n",
      "|               date|time|province|confirmed|released|deceased|  roll_7_confirmed|\n",
      "+-------------------+----+--------+---------+--------+--------+------------------+\n",
      "|2020-03-10 00:00:00|   0|  Sejong|        8|       0|       0| 3.142857142857143|\n",
      "|2020-03-11 00:00:00|   0|  Sejong|       10|       0|       0| 4.428571428571429|\n",
      "|2020-03-12 00:00:00|   0|  Sejong|       15|       0|       0| 6.428571428571429|\n",
      "|2020-03-13 00:00:00|   0|  Sejong|       32|       0|       0|10.857142857142858|\n",
      "|2020-03-14 00:00:00|   0|  Sejong|       38|       0|       0|              16.0|\n",
      "|2020-03-15 00:00:00|   0|  Sejong|       39|       0|       0|21.142857142857142|\n",
      "|2020-03-16 00:00:00|   0|  Sejong|       40|       0|       0|              26.0|\n",
      "|2020-03-17 00:00:00|   0|  Sejong|       40|       0|       0|30.571428571428573|\n",
      "|2020-03-18 00:00:00|   0|  Sejong|       41|       0|       0|              35.0|\n",
      "|2020-03-19 00:00:00|   0|  Sejong|       41|       0|       0|38.714285714285715|\n",
      "|2020-03-20 00:00:00|   0|  Sejong|       41|       0|       0|              40.0|\n",
      "|2020-03-21 00:00:00|   0|  Sejong|       41|       2|       0| 40.42857142857143|\n",
      "|2020-03-22 00:00:00|   0|  Sejong|       41|       3|       0|40.714285714285715|\n",
      "|2020-03-23 00:00:00|   0|  Sejong|       42|       3|       0|              41.0|\n",
      "|2020-03-24 00:00:00|   0|  Sejong|       42|       3|       0|41.285714285714285|\n",
      "|2020-03-25 00:00:00|   0|  Sejong|       44|       3|       0|41.714285714285715|\n",
      "|2020-03-26 00:00:00|   0|  Sejong|       44|       8|       0|42.142857142857146|\n",
      "|2020-03-27 00:00:00|   0|  Sejong|       44|       9|       0| 42.57142857142857|\n",
      "|2020-03-28 00:00:00|   0|  Sejong|       44|       9|       0|              43.0|\n",
      "|2020-03-29 00:00:00|   0|  Sejong|       46|      11|       0|43.714285714285715|\n",
      "+-------------------+----+--------+---------+--------+--------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.window import Window\n",
    "\n",
    "# we only look at the past 7 days in a particular window including the current_day. \n",
    "# Here 0 specifies the current_row and -6 specifies the seventh row previous to current_row. \n",
    "# Remember we count starting from 0.\n",
    "\n",
    "# If we had used rowsBetween(-7,-1), we would just have looked at past 7 days of data and not the current_day\n",
    "windowSpec = Window().partitionBy(['province']).orderBy('date').rowsBetween(-6,0)\n",
    "\n",
    "timeprovinceWithRoll = timeprovince.withColumn(\"roll_7_confirmed\",F.mean(\"confirmed\").over(windowSpec))\n",
    "\n",
    "timeprovinceWithRoll.filter(timeprovinceWithLag.date>'2020-03-10').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "One could also find a use for **rowsBetween(Window.unboundedPreceding, Window.currentRow)** function, where we take the rows between the first row in a window and the current_row to get running totals. I am calculating cumulative_confirmed here."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----+--------+---------+--------+--------+--------------------+\n",
      "|               date|time|province|confirmed|released|deceased|cumulative_confirmed|\n",
      "+-------------------+----+--------+---------+--------+--------+--------------------+\n",
      "|2020-03-10 00:00:00|   0|  Sejong|        8|       0|       0|                  33|\n",
      "|2020-03-11 00:00:00|   0|  Sejong|       10|       0|       0|                  43|\n",
      "|2020-03-12 00:00:00|   0|  Sejong|       15|       0|       0|                  58|\n",
      "|2020-03-13 00:00:00|   0|  Sejong|       32|       0|       0|                  90|\n",
      "|2020-03-14 00:00:00|   0|  Sejong|       38|       0|       0|                 128|\n",
      "|2020-03-15 00:00:00|   0|  Sejong|       39|       0|       0|                 167|\n",
      "|2020-03-16 00:00:00|   0|  Sejong|       40|       0|       0|                 207|\n",
      "|2020-03-17 00:00:00|   0|  Sejong|       40|       0|       0|                 247|\n",
      "|2020-03-18 00:00:00|   0|  Sejong|       41|       0|       0|                 288|\n",
      "|2020-03-19 00:00:00|   0|  Sejong|       41|       0|       0|                 329|\n",
      "|2020-03-20 00:00:00|   0|  Sejong|       41|       0|       0|                 370|\n",
      "|2020-03-21 00:00:00|   0|  Sejong|       41|       2|       0|                 411|\n",
      "|2020-03-22 00:00:00|   0|  Sejong|       41|       3|       0|                 452|\n",
      "|2020-03-23 00:00:00|   0|  Sejong|       42|       3|       0|                 494|\n",
      "|2020-03-24 00:00:00|   0|  Sejong|       42|       3|       0|                 536|\n",
      "|2020-03-25 00:00:00|   0|  Sejong|       44|       3|       0|                 580|\n",
      "|2020-03-26 00:00:00|   0|  Sejong|       44|       8|       0|                 624|\n",
      "|2020-03-27 00:00:00|   0|  Sejong|       44|       9|       0|                 668|\n",
      "|2020-03-28 00:00:00|   0|  Sejong|       44|       9|       0|                 712|\n",
      "|2020-03-29 00:00:00|   0|  Sejong|       46|      11|       0|                 758|\n",
      "+-------------------+----+--------+---------+--------+--------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.window import Window\n",
    "\n",
    "windowSpec = Window().partitionBy(['province']).orderBy('date').rowsBetween(Window.unboundedPreceding,Window.currentRow)\n",
    "\n",
    "timeprovinceWithRoll = timeprovince.withColumn(\"cumulative_confirmed\",F.sum(\"confirmed\").over(windowSpec))\n",
    "\n",
    "timeprovinceWithRoll.filter(timeprovinceWithLag.date>'2020-03-10').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5. Pivot DataFrames"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sometimes we may need to have the dataframe in flat format. This happens frequently in movie data where we may want to show genres as columns instead of rows. We can use pivot to do this. Here I am trying to get one row for each date and getting the province names as columns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>date</th>\n",
       "      <th>Busan_confirmed</th>\n",
       "      <th>Busan_released</th>\n",
       "      <th>Chungcheongbuk-do_confirmed</th>\n",
       "      <th>Chungcheongbuk-do_released</th>\n",
       "      <th>Chungcheongnam-do_confirmed</th>\n",
       "      <th>Chungcheongnam-do_released</th>\n",
       "      <th>Daegu_confirmed</th>\n",
       "      <th>Daegu_released</th>\n",
       "      <th>Daejeon_confirmed</th>\n",
       "      <th>...</th>\n",
       "      <th>Jeollabuk-do_confirmed</th>\n",
       "      <th>Jeollabuk-do_released</th>\n",
       "      <th>Jeollanam-do_confirmed</th>\n",
       "      <th>Jeollanam-do_released</th>\n",
       "      <th>Sejong_confirmed</th>\n",
       "      <th>Sejong_released</th>\n",
       "      <th>Seoul_confirmed</th>\n",
       "      <th>Seoul_released</th>\n",
       "      <th>Ulsan_confirmed</th>\n",
       "      <th>Ulsan_released</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2020-04-10</td>\n",
       "      <td>126</td>\n",
       "      <td>98</td>\n",
       "      <td>45</td>\n",
       "      <td>28</td>\n",
       "      <td>138</td>\n",
       "      <td>105</td>\n",
       "      <td>6807</td>\n",
       "      <td>5193</td>\n",
       "      <td>39</td>\n",
       "      <td>...</td>\n",
       "      <td>17</td>\n",
       "      <td>8</td>\n",
       "      <td>15</td>\n",
       "      <td>4</td>\n",
       "      <td>46</td>\n",
       "      <td>20</td>\n",
       "      <td>595</td>\n",
       "      <td>185</td>\n",
       "      <td>40</td>\n",
       "      <td>28</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2020-03-06</td>\n",
       "      <td>95</td>\n",
       "      <td>2</td>\n",
       "      <td>15</td>\n",
       "      <td>1</td>\n",
       "      <td>90</td>\n",
       "      <td>0</td>\n",
       "      <td>4694</td>\n",
       "      <td>15</td>\n",
       "      <td>18</td>\n",
       "      <td>...</td>\n",
       "      <td>7</td>\n",
       "      <td>2</td>\n",
       "      <td>4</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>105</td>\n",
       "      <td>25</td>\n",
       "      <td>23</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2020-01-20</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2020-04-28</td>\n",
       "      <td>137</td>\n",
       "      <td>116</td>\n",
       "      <td>45</td>\n",
       "      <td>40</td>\n",
       "      <td>142</td>\n",
       "      <td>123</td>\n",
       "      <td>6849</td>\n",
       "      <td>6056</td>\n",
       "      <td>40</td>\n",
       "      <td>...</td>\n",
       "      <td>18</td>\n",
       "      <td>10</td>\n",
       "      <td>15</td>\n",
       "      <td>10</td>\n",
       "      <td>46</td>\n",
       "      <td>35</td>\n",
       "      <td>633</td>\n",
       "      <td>427</td>\n",
       "      <td>43</td>\n",
       "      <td>36</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>2020-02-23</td>\n",
       "      <td>19</td>\n",
       "      <td>0</td>\n",
       "      <td>3</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>310</td>\n",
       "      <td>0</td>\n",
       "      <td>3</td>\n",
       "      <td>...</td>\n",
       "      <td>3</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>31</td>\n",
       "      <td>6</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>2020-03-29</td>\n",
       "      <td>117</td>\n",
       "      <td>85</td>\n",
       "      <td>41</td>\n",
       "      <td>21</td>\n",
       "      <td>127</td>\n",
       "      <td>83</td>\n",
       "      <td>6610</td>\n",
       "      <td>3706</td>\n",
       "      <td>34</td>\n",
       "      <td>...</td>\n",
       "      <td>12</td>\n",
       "      <td>7</td>\n",
       "      <td>9</td>\n",
       "      <td>3</td>\n",
       "      <td>46</td>\n",
       "      <td>11</td>\n",
       "      <td>410</td>\n",
       "      <td>93</td>\n",
       "      <td>39</td>\n",
       "      <td>19</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>2020-04-17</td>\n",
       "      <td>130</td>\n",
       "      <td>112</td>\n",
       "      <td>45</td>\n",
       "      <td>37</td>\n",
       "      <td>139</td>\n",
       "      <td>115</td>\n",
       "      <td>6827</td>\n",
       "      <td>5626</td>\n",
       "      <td>40</td>\n",
       "      <td>...</td>\n",
       "      <td>17</td>\n",
       "      <td>9</td>\n",
       "      <td>15</td>\n",
       "      <td>7</td>\n",
       "      <td>46</td>\n",
       "      <td>24</td>\n",
       "      <td>621</td>\n",
       "      <td>268</td>\n",
       "      <td>42</td>\n",
       "      <td>34</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>2020-06-23</td>\n",
       "      <td>150</td>\n",
       "      <td>142</td>\n",
       "      <td>62</td>\n",
       "      <td>56</td>\n",
       "      <td>161</td>\n",
       "      <td>144</td>\n",
       "      <td>6901</td>\n",
       "      <td>6686</td>\n",
       "      <td>86</td>\n",
       "      <td>...</td>\n",
       "      <td>24</td>\n",
       "      <td>20</td>\n",
       "      <td>20</td>\n",
       "      <td>18</td>\n",
       "      <td>49</td>\n",
       "      <td>47</td>\n",
       "      <td>1230</td>\n",
       "      <td>743</td>\n",
       "      <td>53</td>\n",
       "      <td>48</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>2020-06-15</td>\n",
       "      <td>147</td>\n",
       "      <td>142</td>\n",
       "      <td>61</td>\n",
       "      <td>52</td>\n",
       "      <td>152</td>\n",
       "      <td>144</td>\n",
       "      <td>6894</td>\n",
       "      <td>6667</td>\n",
       "      <td>46</td>\n",
       "      <td>...</td>\n",
       "      <td>21</td>\n",
       "      <td>19</td>\n",
       "      <td>20</td>\n",
       "      <td>18</td>\n",
       "      <td>47</td>\n",
       "      <td>47</td>\n",
       "      <td>1120</td>\n",
       "      <td>689</td>\n",
       "      <td>53</td>\n",
       "      <td>47</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>2020-04-02</td>\n",
       "      <td>122</td>\n",
       "      <td>87</td>\n",
       "      <td>44</td>\n",
       "      <td>23</td>\n",
       "      <td>133</td>\n",
       "      <td>95</td>\n",
       "      <td>6725</td>\n",
       "      <td>4283</td>\n",
       "      <td>36</td>\n",
       "      <td>...</td>\n",
       "      <td>14</td>\n",
       "      <td>7</td>\n",
       "      <td>14</td>\n",
       "      <td>3</td>\n",
       "      <td>46</td>\n",
       "      <td>12</td>\n",
       "      <td>488</td>\n",
       "      <td>127</td>\n",
       "      <td>40</td>\n",
       "      <td>21</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>10 rows × 35 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "        date  Busan_confirmed  Busan_released  Chungcheongbuk-do_confirmed  \\\n",
       "0 2020-04-10              126              98                           45   \n",
       "1 2020-03-06               95               2                           15   \n",
       "2 2020-01-20                0               0                            0   \n",
       "3 2020-04-28              137             116                           45   \n",
       "4 2020-02-23               19               0                            3   \n",
       "5 2020-03-29              117              85                           41   \n",
       "6 2020-04-17              130             112                           45   \n",
       "7 2020-06-23              150             142                           62   \n",
       "8 2020-06-15              147             142                           61   \n",
       "9 2020-04-02              122              87                           44   \n",
       "\n",
       "   Chungcheongbuk-do_released  Chungcheongnam-do_confirmed  \\\n",
       "0                          28                          138   \n",
       "1                           1                           90   \n",
       "2                           0                            0   \n",
       "3                          40                          142   \n",
       "4                           0                            1   \n",
       "5                          21                          127   \n",
       "6                          37                          139   \n",
       "7                          56                          161   \n",
       "8                          52                          152   \n",
       "9                          23                          133   \n",
       "\n",
       "   Chungcheongnam-do_released  Daegu_confirmed  Daegu_released  \\\n",
       "0                         105             6807            5193   \n",
       "1                           0             4694              15   \n",
       "2                           0                0               0   \n",
       "3                         123             6849            6056   \n",
       "4                           0              310               0   \n",
       "5                          83             6610            3706   \n",
       "6                         115             6827            5626   \n",
       "7                         144             6901            6686   \n",
       "8                         144             6894            6667   \n",
       "9                          95             6725            4283   \n",
       "\n",
       "   Daejeon_confirmed  ...  Jeollabuk-do_confirmed  Jeollabuk-do_released  \\\n",
       "0                 39  ...                      17                      8   \n",
       "1                 18  ...                       7                      2   \n",
       "2                  0  ...                       0                      0   \n",
       "3                 40  ...                      18                     10   \n",
       "4                  3  ...                       3                      1   \n",
       "5                 34  ...                      12                      7   \n",
       "6                 40  ...                      17                      9   \n",
       "7                 86  ...                      24                     20   \n",
       "8                 46  ...                      21                     19   \n",
       "9                 36  ...                      14                      7   \n",
       "\n",
       "   Jeollanam-do_confirmed  Jeollanam-do_released  Sejong_confirmed  \\\n",
       "0                      15                      4                46   \n",
       "1                       4                      0                 1   \n",
       "2                       0                      0                 0   \n",
       "3                      15                     10                46   \n",
       "4                       1                      0                 1   \n",
       "5                       9                      3                46   \n",
       "6                      15                      7                46   \n",
       "7                      20                     18                49   \n",
       "8                      20                     18                47   \n",
       "9                      14                      3                46   \n",
       "\n",
       "   Sejong_released  Seoul_confirmed  Seoul_released  Ulsan_confirmed  \\\n",
       "0               20              595             185               40   \n",
       "1                0              105              25               23   \n",
       "2                0                0               0                0   \n",
       "3               35              633             427               43   \n",
       "4                0               31               6                1   \n",
       "5               11              410              93               39   \n",
       "6               24              621             268               42   \n",
       "7               47             1230             743               53   \n",
       "8               47             1120             689               53   \n",
       "9               12              488             127               40   \n",
       "\n",
       "   Ulsan_released  \n",
       "0              28  \n",
       "1               1  \n",
       "2               0  \n",
       "3              36  \n",
       "4               0  \n",
       "5              19  \n",
       "6              34  \n",
       "7              48  \n",
       "8              47  \n",
       "9              21  \n",
       "\n",
       "[10 rows x 35 columns]"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pivotedTimeprovince = timeprovince.groupBy('date').pivot('province') \\\n",
    ".agg(F.sum('confirmed').alias('confirmed') , F.sum('released').alias('released'))\n",
    "\n",
    "pivotedTimeprovince.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 6. Other Opertions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [1] Caching"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark works on the lazy execution principle. What that means is that nothing really gets executed until you use an action function like the .count() on a dataframe. And if you do a .count function, it generally helps to cache at this step. So I have made it a point to cache() my dataframes whenever I do a .count() operation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.cache().count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [2] Save and Load from an intermediate step"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When you work with Spark you will frequently run with memory and storage issues. While in some cases such issues might be resolved using techniques like broadcasting, salting or cache, sometimes just interrupting the workflow and saving and reloading the whole dataframe at a crucial step has helped me a lot. This helps spark to let go of a lot of memory that gets utilized for storing intermediate shuffle data and unused caches."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.write.parquet(\"data/df.parquet\")\n",
    "df.unpersist()\n",
    "spark.read.load(\"data/df.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [3] Repartitioning"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You might want to repartition your data if you feel your data has been skewed while working with all the transformations and joins. The simplest way to do it is by using:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.repartition(1000)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sometimes you might also want to repartition by a known scheme as this scheme might be used by a certain join or aggregation operation later on. You can use multiple columns to repartition using:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.repartition('cola', 'colb','colc','cold')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then, we can get the number of partitions in a data frame using:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also check out the distribution of records in a partition by using the glom function. This helps in understanding the skew in the data that happens while working with various transformations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.glom().map(len).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [4] Reading Parquet File in Local\n",
    "Sometimes you might want to read the parquet files in a system where Spark is not available. In such cases, I normally use the below code:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from glob import glob\n",
    "def load_df_from_parquet(parquet_directory):\n",
    "    df = pd.DataFrame()\n",
    "    for file in glob(f\"{parquet_directory}/*\"):\n",
    "        df = pd.concat([df,pd.read_parquet(file)])\n",
    "    return df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 7. Close Spark Instance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Reference:\n",
    ">https://medium.com/@rahul_agarwal"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}