EDA_008_local_analysis_par_i94_by_i94yr_i94mon_v2.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Objective: load parititioned (by i94yr and i94mon) i94 dataset, where \n",
    "\n",
    "* Attempt 1: where ... `i94yr=2016 and i94mon=3`.  (successful)\n",
    "* Attempt 2: where ... `i94yr=2016 and (i94mon between 3 an 6)`.  (successful)\n",
    "\n",
    "Lesson learnt: turns out the partitionBy variables in the partitioned dataset remain in the dataset, if we read it correctly!!\n",
    "\n",
    "Open ended question. Which design is better?\n",
    "\n",
    "\n",
    "* Attempt 1: \n",
    "    * where ... `i94yr=2016 and i94mon=3`, or\n",
    "    * where ... `ym=201603`\n",
    "* Attempt 2: `i94yr=2016 and (i94mon between 3 an 6)`\n",
    "    * where ... `i94yr=2016 and (i94mon between 3 and 9)`, or\n",
    "    * where ... `ym between 201603 and 201609`\n",
    "    \n",
    "Either options have its pros and cons. Bottom line is, I believe both options work.\n",
    "\n",
    "Paritioning by `i94yr` and `i94mon` has the benefit of easier querying for just year, or just month. But if we want data from across multiple years, the where query may be longer.\n",
    "\n",
    "Paritioning by (a newly created `ym` has the effectively opposite pros/cons of the above."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import pyspark\n",
    "import configparser\n",
    "import pandas as pd\n",
    "import pyspark.sql.functions as F\n",
    "import pyspark.sql.types as T\n",
    "\n",
    "from datetime import datetime, timedelta\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Ensure Jupyter Notebook display pandas dataframe fully. Show all columns. Do not truncate column value.\n",
    "pd.set_option('display.max_columns', None)\n",
    "pd.set_option('display.max_colwidth', 1000)\n",
    "\n",
    "# Dev config from the non-secret configuration file\n",
    "config_dev = configparser.ConfigParser()\n",
    "config_dev.read_file(open('aws_dev.cfg'))\n",
    "\n",
    "PAR_I94_FILE_BY_I94YR_I94MON = config_dev.get('DATA_PATHS_LOCAL', 'PAR_I94_FILE_BY_I94YR_I94MON')\n",
    "PAR_I94_DIR_BY_I94YR_I94MON = config_dev.get('DATA_PATHS_LOCAL', 'PAR_I94_DIR_BY_I94YR_I94MON')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.\\\n",
    "    config(\"spark.jars.repositories\", \"https://repos.spark-packages.org/\").\\\n",
    "    config(\"spark.jars.packages\", \"saurfang:spark-sas7bdat:2.0.0-s_2.11\").\\\n",
    "    enableHiveSupport().getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "https://stackoverflow.com/questions/47191078/spark-sql-queries-on-partitioned-data-using-date-ranges\n",
    "\n",
    "```\n",
    "spark.read.parquet(\"hdfs:///basepath\")\n",
    "  .where('ts >= 201710060000L && 'ts <= 201711030000L)\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Attempt 1: read one particular year-month (partition)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_i94 = spark.read.parquet(PAR_I94_DIR_BY_I94YR_I94MON)\\\n",
    "    .where('i94yr=2016 and i94mon=3')\n",
    "\n",
    "# This works too\n",
    "# df_i94 = spark.read.parquet(PAR_I94_DIR_BY_I94YR_I94MON)\\\n",
    "#     .where('i94yr=2016.0 and i94mon=3.0')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- cicid: double (nullable = true)\n",
      " |-- i94cit: double (nullable = true)\n",
      " |-- i94res: double (nullable = true)\n",
      " |-- i94port: string (nullable = true)\n",
      " |-- arrdate: double (nullable = true)\n",
      " |-- i94mode: double (nullable = true)\n",
      " |-- i94addr: string (nullable = true)\n",
      " |-- depdate: double (nullable = true)\n",
      " |-- i94bir: double (nullable = true)\n",
      " |-- i94visa: double (nullable = true)\n",
      " |-- count: double (nullable = true)\n",
      " |-- dtadfile: string (nullable = true)\n",
      " |-- visapost: string (nullable = true)\n",
      " |-- occup: string (nullable = true)\n",
      " |-- entdepa: string (nullable = true)\n",
      " |-- entdepd: string (nullable = true)\n",
      " |-- entdepu: string (nullable = true)\n",
      " |-- matflag: string (nullable = true)\n",
      " |-- biryear: double (nullable = true)\n",
      " |-- dtaddto: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- insnum: string (nullable = true)\n",
      " |-- airline: string (nullable = true)\n",
      " |-- admnum: double (nullable = true)\n",
      " |-- fltno: string (nullable = true)\n",
      " |-- visatype: string (nullable = true)\n",
      " |-- i94yr: double (nullable = true)\n",
      " |-- i94mon: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_i94.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "def convert_datetime(sas_date):\n",
    "    try:\n",
    "        if (sas_date == 'null'):\n",
    "            sas_date = 0\n",
    "        start_cutoff = datetime(1960, 1, 1)\n",
    "        return start_cutoff + timedelta(days=int(sas_date))\n",
    "    except:\n",
    "        return None\n",
    "    \n",
    "udf_datetime_from_sas = F.udf(lambda x: convert_datetime(x), T.DateType())\n",
    "df_i94 = df_i94.withColumn('arrdate', udf_datetime_from_sas(df_i94.arrdate))\n",
    "df_i94 = df_i94.withColumn('depdate', udf_datetime_from_sas(df_i94.depdate))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_i94.createOrReplaceTempView('df_i94')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-------------+--------+\n",
      "|year_arrdate|month_arrdate|count(1)|\n",
      "+------------+-------------+--------+\n",
      "|        2016|            3| 3157072|\n",
      "+------------+-------------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "select\n",
    "    year(arrdate) as year_arrdate,\n",
    "    month(arrdate) as month_arrdate,\n",
    "    count(*)\n",
    "from df_i94\n",
    "group by 1, 2\n",
    "order by 1, 2\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+--------+\n",
      "| i94yr|i94mon|count(1)|\n",
      "+------+------+--------+\n",
      "|2016.0|   3.0| 3157072|\n",
      "+------+------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "select\n",
    "    i94yr,\n",
    "    i94mon,\n",
    "    count(*)\n",
    "from df_i94\n",
    "group by 1, 2\n",
    "order by 1, 2\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Attemp 2: read a range of months"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_i94 = spark.read.parquet(PAR_I94_DIR_BY_I94YR_I94MON)\\\n",
    "    .where('i94yr=2016 and (i94mon between 3 and 6)')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_i94 = df_i94.withColumn('arrdate', udf_datetime_from_sas(df_i94.arrdate))\n",
    "df_i94 = df_i94.withColumn('depdate', udf_datetime_from_sas(df_i94.depdate))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_i94.createOrReplaceTempView('df_i94')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-------------+--------+\n",
      "|year_arrdate|month_arrdate|count(1)|\n",
      "+------------+-------------+--------+\n",
      "|        2016|            3| 3157072|\n",
      "|        2016|            4| 3096313|\n",
      "|        2016|            5| 3444249|\n",
      "|        2016|            6| 3574989|\n",
      "+------------+-------------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "select\n",
    "    year(arrdate) as year_arrdate,\n",
    "    month(arrdate) as month_arrdate,\n",
    "    count(*)\n",
    "from df_i94\n",
    "group by 1, 2\n",
    "order by 1, 2\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+--------+\n",
      "| i94yr|i94mon|count(1)|\n",
      "+------+------+--------+\n",
      "|2016.0|   3.0| 3157072|\n",
      "|2016.0|   4.0| 3096313|\n",
      "|2016.0|   5.0| 3444249|\n",
      "|2016.0|   6.0| 3574989|\n",
      "+------+------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "select\n",
    "    i94yr,\n",
    "    i94mon,\n",
    "    count(*)\n",
    "from df_i94\n",
    "group by 1, 2\n",
    "order by 1, 2\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "editable": true
   },
   "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.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}