SparkR-Loading-CSV.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Loading CSV Data Files with SparkR\n",
    "This is the \"Hello, World!\" for loading data from CSV files into Spark.\n",
    "\n",
    "This example is specific for the configuration of the Institute for Insight Hadoop/Spark cluster (as of Fall 2016). We're currently running Spark 1.6.1.\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "    $ hdfs dfs -ls /user/mgrace/red_cross/\n",
    "    Found 3 items\n",
    "    -rw-r--r--   3 mgrace hadoop   142603317 2016-09-02 10:24 /user/mgrace/red_cross/donor_deferral912016.csv\n",
    "    -rw-r--r--   3 mgrace hadoop  4763358313 2016-09-02 10:24 /user/mgrace/red_cross/donor_summary912016.csv\n",
    "    -rw-r--r--   3 mgrace hadoop 10551624919 2016-09-02 10:26 /user/mgrace/red_cross/donor_transactions912016.csv"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Initialize Spark\n",
    "The following lines load the \"SparkR\" library, start a Spark session, and initialize the SQL context. If this step needs to be repeated a current Spark session should be shut-down first. These commands can also be exectued in RStudio.\n",
    "\n",
    "**References**:\n",
    "- https://github.com/jadianes/spark-r-notebooks/blob/master/notebooks/nb0-starting-up/nb0-starting-up.ipynb\n",
    "- https://spark.apache.org/docs/1.6.1/api/java/org/apache/spark/sql/functions.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n",
      "Attaching package: ‘SparkR’\n",
      "\n",
      "The following objects are masked from ‘package:stats’:\n",
      "\n",
      "    cov, filter, lag, na.omit, predict, sd, var\n",
      "\n",
      "The following objects are masked from ‘package:base’:\n",
      "\n",
      "    colnames, colnames<-, endsWith, intersect, rank, rbind, sample,\n",
      "    startsWith, subset, summary, table, transform\n",
      "\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Launching java with spark-submit command /usr/hdp/2.4.2.0-258/spark/bin/spark-submit  --packages com.databricks:spark-csv_2.11:1.2.0 sparkr-shell /tmp/Rtmpy8cjYz/backend_port18905263a7f5a \n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Loading library\n",
    "#\n",
    "Sys.setenv(SPARK_HOME='/usr/hdp/2.4.2.0-258/spark')\n",
    ".libPaths(c(file.path(Sys.getenv('SPARK_HOME'), 'R', 'lib'), .libPaths()))\n",
    "library(SparkR)\n",
    "#\n",
    "# Starting SparkR session (run `sparkR.stop()` before starting a new session)\n",
    "#\n",
    "#sparkR.stop()    ### this command will stop the running Spark session\n",
    "sc <- sparkR.init(sparkPackages = \"com.databricks:spark-csv_2.11:1.2.0\")\n",
    "sqlctx <- sparkRSQL.init(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Reading Data Files\n",
    "Let's start with the smallest (and simplest) data file `/user/mgrace/red_cross/donor_deferral912016.csv`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType\n",
       "|-name = \"arc_id\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"deferral_start_date\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"deferral_end_date\", type = \"StringType\", nullable = TRUE"
      ]
     },
     "execution_count": 108,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_deferral <- read.df(sqlctx, \"/user/mgrace/red_cross/donor_deferral912016.csv\", \"com.databricks.spark.csv\", header=\"true\")\n",
    "schema(df_deferral)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "All fields are of type `StringType`. The 'spark-csv' package is able to infer the table schema from the data itself. However, this can be fairly time consuming, and may not always suffice."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 131,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType\n",
       "|-name = \"arc_id\", type = \"IntegerType\", nullable = TRUE\n",
       "|-name = \"deferral_start_date\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"deferral_end_date\", type = \"StringType\", nullable = TRUE"
      ]
     },
     "execution_count": 131,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_deferral <- read.df(sqlctx, \"/user/mgrace/red_cross/donor_deferral912016.csv\", \"com.databricks.spark.csv\", header=\"true\", inferSchema=\"true\")\n",
    "schema(df_deferral)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The columns `deferral_start_date` and `deferral_end_date` are still strings, even though we see that they hold dates. The alterntive is to define custom schema."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define a Custom Schema\n",
    "The following example shows how to create a custom schema for a table and apply it to the loading process. We define the first column as \"integer\" and the other two as \"date\". In this step we may also change the names of the columns. The names in the custom schema override those headers in the CSV file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "deferralSchema <- structType(\n",
    "    structField(\"arc_id\", \"integer\", nullable = TRUE),\n",
    "    structField(\"deferral_start_date\", \"date\", nullable = TRUE),\n",
    "    structField(\"deferral_end_date\", \"date\", nullable = TRUE))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType\n",
       "|-name = \"arc_id\", type = \"IntegerType\", nullable = TRUE\n",
       "|-name = \"deferral_start_date\", type = \"DateType\", nullable = TRUE\n",
       "|-name = \"deferral_end_date\", type = \"DateType\", nullable = TRUE"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_deferral <- read.df(sqlctx, \"/user/mgrace/red_cross/donor_deferral912016.csv\", \"com.databricks.spark.csv\", header=\"true\", schema=deferralSchema)\n",
    "schema(df_deferral)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Everything looks good. Unfortunately it's not. Because Spark uses a *lazy evaluation* sheme nothing has been executed, yet. Once we actually call on getting some data we will learn that our custom schema fails."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 135,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "ename": "ERROR",
     "evalue": "Error in invokeJava(isStatic = TRUE, className, methodName, ...): org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 47.0 failed 1 times, most recent failure: ...\n",
     "output_type": "error",
     "traceback": [
      "Error in invokeJava(isStatic = TRUE, className, methodName, ...): org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 47.0 failed 1 times, most recent failure: ...\n"
     ]
    }
   ],
   "source": [
    "##\n",
    "## Execute the command below\n",
    "##\n",
    "#head(df_deferral)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We have to take ***one step back***. Our date columns are not read properly because of to the format \"yyyy/mm/dd\" in which they were stored. We need to resort back to strings.\n",
    "\n",
    "We give the two date columns diffent names by preceding them with `\"tmp\"`. How to use the renamed columns and add columns of the original table with the proper data type will be shown in the next section."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "deferralSchema_2 <- structType(\n",
    "    structField(\"arc_id\", \"integer\", nullable = TRUE),\n",
    "    structField(\"tmp_deferral_start_date\", \"string\", nullable = TRUE),\n",
    "    structField(\"tmp_deferral_end_date\", \"string\", nullable = TRUE))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType\n",
       "|-name = \"arc_id\", type = \"IntegerType\", nullable = TRUE\n",
       "|-name = \"tmp_deferral_start_date\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"tmp_deferral_end_date\", type = \"StringType\", nullable = TRUE"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_deferral <- read.df(sqlctx, \"/user/mgrace/red_cross/donor_deferral912016.csv\", \"com.databricks.spark.csv\", header=\"true\",\n",
    "                       schema=deferralSchema_2)\n",
    "schema(df_deferral)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<thead><tr><th></th><th scope=col>arc_id</th><th scope=col>tmp_deferral_start_date</th><th scope=col>tmp_deferral_end_date</th></tr></thead>\n",
       "<tbody>\n",
       "\t<tr><th scope=row>1</th><td>60391835  </td><td>2002/06/13</td><td>2003/06/14</td></tr>\n",
       "\t<tr><th scope=row>2</th><td>77352072  </td><td>2007/04/21</td><td>2008/04/22</td></tr>\n",
       "\t<tr><th scope=row>3</th><td>59320300  </td><td>2001/11/07</td><td>2003/06/01</td></tr>\n",
       "\t<tr><th scope=row>4</th><td>70083734  </td><td>2000/10/05</td><td>2000/10/18</td></tr>\n",
       "\t<tr><th scope=row>5</th><td>50383290  </td><td>1997/07/09</td><td>1997/07/15</td></tr>\n",
       "\t<tr><th scope=row>6</th><td>67244463  </td><td>2001/11/12</td><td>2222/02/02</td></tr>\n",
       "</tbody>\n",
       "</table>\n"
      ],
      "text/latex": [
       "\\begin{tabular}{r|lll}\n",
       "  & arc\\_id & tmp\\_deferral\\_start\\_date & tmp\\_deferral\\_end\\_date\\\\\n",
       "\\hline\n",
       "\t1 & 60391835   & 2002/06/13 & 2003/06/14\\\\\n",
       "\t2 & 77352072   & 2007/04/21 & 2008/04/22\\\\\n",
       "\t3 & 59320300   & 2001/11/07 & 2003/06/01\\\\\n",
       "\t4 & 70083734   & 2000/10/05 & 2000/10/18\\\\\n",
       "\t5 & 50383290   & 1997/07/09 & 1997/07/15\\\\\n",
       "\t6 & 67244463   & 2001/11/12 & 2222/02/02\\\\\n",
       "\\end{tabular}\n"
      ],
      "text/plain": [
       "    arc_id tmp_deferral_start_date tmp_deferral_end_date\n",
       "1 60391835              2002/06/13            2003/06/14\n",
       "2 77352072              2007/04/21            2008/04/22\n",
       "3 59320300              2001/11/07            2003/06/01\n",
       "4 70083734              2000/10/05            2000/10/18\n",
       "5 50383290              1997/07/09            1997/07/15\n",
       "6 67244463              2001/11/12            2222/02/02"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "head(df_deferral)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's look at the next data file:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType\n",
       "|-name = \"bzd_assessedhomevalue\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_avg_bank_credit6\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_avg_inq_all12\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_avg_maxcred_install6\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_avg_mos_autopay\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_avg_numauto12\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_descretionary_spend\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_income\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_lengthofresidence\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_mortg_equity\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_numberadultsinhh\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_numberofchildrenhh\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_realty_mortgremain\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_realty_mospay\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_tt_all_in_name\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_tt_buy_am\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_tt_go_flow\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_tt_look_now\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_tt_nsueh\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_tt_penny\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_tt_road_again\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_tt_show_money\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_tt_smell_roses\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_tt_time_like_pres\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_tt_whph\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_affinity_ind\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_aff_demparty\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_aff_ind\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_aff_republican\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cause_animal\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cause_children\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cause_env\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cause_health\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cause_polcons\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cause_pollib\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cause_rel\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cause_vet\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cause_volunt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cont_animal\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cont_childwel\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cont_conspoli\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cont_culture\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cont_env\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cont_health\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cont_humanitarian\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cont_liberal\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cont_pol\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cont_relcont\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_cont_social\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_fin_ira\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_fin_lifeins\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_fin_monetmkt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_fin_mutual\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_fin_real\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_fin_stock\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_inactivemil\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_activemil\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_int_affluence\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_int_politics\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_bb_int_relig\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_childrenage0_18\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_homeowner\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_householdoccupation1\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_maritalstatus1\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_mosaic_hh\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_occ_model1\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_person_type1\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_political\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_religion\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"bzd_education\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"blood_type\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"no_show_cnt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"last_no_show_dt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"fst_no_show_dt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"arc_id\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"birth_dt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"fr_fst_dntn_dt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"fr_last_dntn_dt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"fr_lftm_dntn_amt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"fr_lftm_dntn_cnt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"fr_lftm_max_dntn_amt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"fr_lftm_min_dntn_amt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"lftm_crs_complt_cnt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"fst_crs_cmptn_dt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"last_crs_cmptn_dt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"lftm_hrs_vol_cnt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"last_vol_dt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"first_vol_dt\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"zip5\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"zip4\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"gender\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"race\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"zip5c\", type = \"StringType\", nullable = TRUE"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_summary <- read.df(sqlctx, \"/user/mgrace/red_cross/donor_summary912016.csv\", \"com.databricks.spark.csv\", header=\"true\", nullValue='')\n",
    "schema(df_summary)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The summary table has 93 columns. Instead of creating a custom schema for this table we may just use the default `StringType` and convert selected columns as we need them. This is demonstrated in the next section."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Adding Columns, Fixing Dates\n",
    "Our data files use the format `yyyy/mm/dd`. If we where to use the type `\"date\"` in the schema our load procedure would fail. Therefore, we read data fields as strings into temporary columns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType\n",
       "|-name = \"arc_id\", type = \"IntegerType\", nullable = TRUE\n",
       "|-name = \"tmp_deferral_start_date\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"tmp_deferral_end_date\", type = \"StringType\", nullable = TRUE"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "schema(df_deferral)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<thead><tr><th></th><th scope=col>arc_id</th><th scope=col>tmp_deferral_start_date</th><th scope=col>tmp_deferral_end_date</th></tr></thead>\n",
       "<tbody>\n",
       "\t<tr><th scope=row>1</th><td>60391835  </td><td>2002/06/13</td><td>2003/06/14</td></tr>\n",
       "\t<tr><th scope=row>2</th><td>77352072  </td><td>2007/04/21</td><td>2008/04/22</td></tr>\n",
       "\t<tr><th scope=row>3</th><td>59320300  </td><td>2001/11/07</td><td>2003/06/01</td></tr>\n",
       "\t<tr><th scope=row>4</th><td>70083734  </td><td>2000/10/05</td><td>2000/10/18</td></tr>\n",
       "\t<tr><th scope=row>5</th><td>50383290  </td><td>1997/07/09</td><td>1997/07/15</td></tr>\n",
       "\t<tr><th scope=row>6</th><td>67244463  </td><td>2001/11/12</td><td>2222/02/02</td></tr>\n",
       "</tbody>\n",
       "</table>\n"
      ],
      "text/latex": [
       "\\begin{tabular}{r|lll}\n",
       "  & arc\\_id & tmp\\_deferral\\_start\\_date & tmp\\_deferral\\_end\\_date\\\\\n",
       "\\hline\n",
       "\t1 & 60391835   & 2002/06/13 & 2003/06/14\\\\\n",
       "\t2 & 77352072   & 2007/04/21 & 2008/04/22\\\\\n",
       "\t3 & 59320300   & 2001/11/07 & 2003/06/01\\\\\n",
       "\t4 & 70083734   & 2000/10/05 & 2000/10/18\\\\\n",
       "\t5 & 50383290   & 1997/07/09 & 1997/07/15\\\\\n",
       "\t6 & 67244463   & 2001/11/12 & 2222/02/02\\\\\n",
       "\\end{tabular}\n"
      ],
      "text/plain": [
       "    arc_id tmp_deferral_start_date tmp_deferral_end_date\n",
       "1 60391835              2002/06/13            2003/06/14\n",
       "2 77352072              2007/04/21            2008/04/22\n",
       "3 59320300              2001/11/07            2003/06/01\n",
       "4 70083734              2000/10/05            2000/10/18\n",
       "5 50383290              1997/07/09            1997/07/15\n",
       "6 67244463              2001/11/12            2222/02/02"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "head(df_deferral)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can, now, use the `selectExpr` method to create a new data frame with some of the original columns, and some that are calculated.\n",
    "Expresssions are of the same form as the ones in \"SELECT\" statements in SparkSQL. A list of available functions can be found at https://spark.apache.org/docs/1.6.1/api/java/org/apache/spark/sql/functions.html\n",
    "\n",
    "The first argument of `selectExpr` is a (Spark) data frame. The remaining arguments are strings of expressions, whereby the simplest expression would be just a column name.\n",
    "\n",
    "Transforming data frames into new once with computed columns is common practice in Spark. The transformation aren't actually performed until we aggregate or collect rows. Spark uses *lazy evaluation*."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType\n",
       "|-name = \"arc_id\", type = \"IntegerType\", nullable = TRUE\n",
       "|-name = \"deferral_start_ts\", type = \"LongType\", nullable = TRUE\n",
       "|-name = \"deferral_start_string\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"deferral_start_date\", type = \"DateType\", nullable = TRUE\n",
       "|-name = \"deferral_end_ts\", type = \"LongType\", nullable = TRUE\n",
       "|-name = \"deferral_end_string\", type = \"StringType\", nullable = TRUE\n",
       "|-name = \"deferral_end_date\", type = \"DateType\", nullable = TRUE"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_deferral2 <- selectExpr(df_deferral, \"arc_id\"\n",
    "                ,\"unix_timestamp(tmp_deferral_start_date, 'yyyy/MM/dd') as deferral_start_ts\"\n",
    "                ,\"from_unixtime(unix_timestamp(tmp_deferral_start_date, 'yyyy/MM/dd')) as deferral_start_string\"\n",
    "                ,\"to_date(from_unixtime(unix_timestamp(tmp_deferral_start_date, 'yyyy/MM/dd'))) as deferral_start_date\"\n",
    "                ,\"unix_timestamp(tmp_deferral_end_date, 'yyyy/MM/dd') as deferral_end_ts\"\n",
    "                ,\"from_unixtime(unix_timestamp(tmp_deferral_end_date, 'yyyy/MM/dd')) as deferral_end_string\"\n",
    "                ,\"to_date(from_unixtime(unix_timestamp(tmp_deferral_end_date, 'yyyy/MM/dd'))) as deferral_end_date\"\n",
    "               )\n",
    "schema(df_deferral2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<thead><tr><th></th><th scope=col>arc_id</th><th scope=col>deferral_start_ts</th><th scope=col>deferral_start_string</th><th scope=col>deferral_start_date</th><th scope=col>deferral_end_ts</th><th scope=col>deferral_end_string</th><th scope=col>deferral_end_date</th></tr></thead>\n",
       "<tbody>\n",
       "\t<tr><th scope=row>1</th><td>60391835           </td><td>1023940800         </td><td>2002-06-13 00:00:00</td><td>11851              </td><td>1055563200         </td><td>2003-06-14 00:00:00</td><td>12217              </td></tr>\n",
       "\t<tr><th scope=row>2</th><td>77352072           </td><td>1177128000         </td><td>2007-04-21 00:00:00</td><td>13624              </td><td>1208836800         </td><td>2008-04-22 00:00:00</td><td>13991              </td></tr>\n",
       "\t<tr><th scope=row>3</th><td>59320300           </td><td>1005109200         </td><td>2001-11-07 00:00:00</td><td>11633              </td><td>1054440000         </td><td>2003-06-01 00:00:00</td><td>12204              </td></tr>\n",
       "\t<tr><th scope=row>4</th><td>70083734           </td><td>970718400          </td><td>2000-10-05 00:00:00</td><td>11235              </td><td>971841600          </td><td>2000-10-18 00:00:00</td><td>11248              </td></tr>\n",
       "\t<tr><th scope=row>5</th><td>50383290           </td><td>868420800          </td><td>1997-07-09 00:00:00</td><td>10051              </td><td>868939200          </td><td>1997-07-15 00:00:00</td><td>10057              </td></tr>\n",
       "\t<tr><th scope=row>6</th><td>67244463           </td><td>1005541200         </td><td>2001-11-12 00:00:00</td><td>11638              </td><td>7955125200         </td><td>2222-02-02 00:00:00</td><td>92073              </td></tr>\n",
       "</tbody>\n",
       "</table>\n"
      ],
      "text/latex": [
       "\\begin{tabular}{r|lllllll}\n",
       "  & arc\\_id & deferral\\_start\\_ts & deferral\\_start\\_string & deferral\\_start\\_date & deferral\\_end\\_ts & deferral\\_end\\_string & deferral\\_end\\_date\\\\\n",
       "\\hline\n",
       "\t1 & 60391835            & 1023940800          & 2002-06-13 00:00:00 & 11851               & 1055563200          & 2003-06-14 00:00:00 & 12217              \\\\\n",
       "\t2 & 77352072            & 1177128000          & 2007-04-21 00:00:00 & 13624               & 1208836800          & 2008-04-22 00:00:00 & 13991              \\\\\n",
       "\t3 & 59320300            & 1005109200          & 2001-11-07 00:00:00 & 11633               & 1054440000          & 2003-06-01 00:00:00 & 12204              \\\\\n",
       "\t4 & 70083734            & 970718400           & 2000-10-05 00:00:00 & 11235               & 971841600           & 2000-10-18 00:00:00 & 11248              \\\\\n",
       "\t5 & 50383290            & 868420800           & 1997-07-09 00:00:00 & 10051               & 868939200           & 1997-07-15 00:00:00 & 10057              \\\\\n",
       "\t6 & 67244463            & 1005541200          & 2001-11-12 00:00:00 & 11638               & 7955125200          & 2222-02-02 00:00:00 & 92073              \\\\\n",
       "\\end{tabular}\n"
      ],
      "text/plain": [
       "    arc_id deferral_start_ts deferral_start_string deferral_start_date\n",
       "1 60391835        1023940800   2002-06-13 00:00:00          2002-06-13\n",
       "2 77352072        1177128000   2007-04-21 00:00:00          2007-04-21\n",
       "3 59320300        1005109200   2001-11-07 00:00:00          2001-11-07\n",
       "4 70083734         970718400   2000-10-05 00:00:00          2000-10-05\n",
       "5 50383290         868420800   1997-07-09 00:00:00          1997-07-09\n",
       "6 67244463        1005541200   2001-11-12 00:00:00          2001-11-12\n",
       "  deferral_end_ts deferral_end_string deferral_end_date\n",
       "1      1055563200 2003-06-14 00:00:00        2003-06-14\n",
       "2      1208836800 2008-04-22 00:00:00        2008-04-22\n",
       "3      1054440000 2003-06-01 00:00:00        2003-06-01\n",
       "4       971841600 2000-10-18 00:00:00        2000-10-18\n",
       "5       868939200 1997-07-15 00:00:00        1997-07-15\n",
       "6      7955125200 2222-02-02 00:00:00        2222-02-02"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "head(df_deferral2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The '\\_ts' columns show the result of the `unix_timestamp` function. The UNIX time is represented as the number of elapsed seconds since midnight, January 1, 1970. This function allows for a format argument; in this case \"yyyy/MM/dd\".\n",
    "\n",
    "Converting unix time back into text format with `from_unixtime` shows the official datetime format that Spark (and Java) use. As seen in the '\\_string' columns.\n",
    "\n",
    "The actual '\\_date' columns also contain numbers, which are not that useful when looking at the data. However, this data format is essential for the many date functions that Spark offers, as the following example demonstrates."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df3 <- selectExpr(df_deferral2, \"arc_id\", \"deferral_start_date\"\n",
    "                 ,\"year(deferral_start_date) as start_year\"\n",
    "                 ,\"month(deferral_start_date) as start_month\"\n",
    "                 ,\"day(deferral_start_date) as start_day\"\n",
    "                 )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType\n",
       "|-name = \"arc_id\", type = \"IntegerType\", nullable = TRUE\n",
       "|-name = \"deferral_start_date\", type = \"DateType\", nullable = TRUE\n",
       "|-name = \"start_year\", type = \"IntegerType\", nullable = TRUE\n",
       "|-name = \"start_month\", type = \"IntegerType\", nullable = TRUE\n",
       "|-name = \"start_day\", type = \"IntegerType\", nullable = TRUE"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "schema(df3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<thead><tr><th></th><th scope=col>arc_id</th><th scope=col>deferral_start_date</th><th scope=col>start_year</th><th scope=col>start_month</th><th scope=col>start_day</th></tr></thead>\n",
       "<tbody>\n",
       "\t<tr><th scope=row>1</th><td>60391835</td><td>   11851</td><td>    2002</td><td>       6</td><td>      13</td></tr>\n",
       "\t<tr><th scope=row>2</th><td>77352072</td><td>   13624</td><td>    2007</td><td>       4</td><td>      21</td></tr>\n",
       "\t<tr><th scope=row>3</th><td>59320300</td><td>   11633</td><td>    2001</td><td>      11</td><td>       7</td></tr>\n",
       "\t<tr><th scope=row>4</th><td>70083734</td><td>   11235</td><td>    2000</td><td>      10</td><td>       5</td></tr>\n",
       "\t<tr><th scope=row>5</th><td>50383290</td><td>   10051</td><td>    1997</td><td>       7</td><td>       9</td></tr>\n",
       "\t<tr><th scope=row>6</th><td>67244463</td><td>   11638</td><td>    2001</td><td>      11</td><td>      12</td></tr>\n",
       "</tbody>\n",
       "</table>\n"
      ],
      "text/latex": [
       "\\begin{tabular}{r|lllll}\n",
       "  & arc\\_id & deferral\\_start\\_date & start\\_year & start\\_month & start\\_day\\\\\n",
       "\\hline\n",
       "\t1 & 60391835 &    11851 &     2002 &        6 &       13\\\\\n",
       "\t2 & 77352072 &    13624 &     2007 &        4 &       21\\\\\n",
       "\t3 & 59320300 &    11633 &     2001 &       11 &        7\\\\\n",
       "\t4 & 70083734 &    11235 &     2000 &       10 &        5\\\\\n",
       "\t5 & 50383290 &    10051 &     1997 &        7 &        9\\\\\n",
       "\t6 & 67244463 &    11638 &     2001 &       11 &       12\\\\\n",
       "\\end{tabular}\n"
      ],
      "text/plain": [
       "    arc_id deferral_start_date start_year start_month start_day\n",
       "1 60391835          2002-06-13       2002           6        13\n",
       "2 77352072          2007-04-21       2007           4        21\n",
       "3 59320300          2001-11-07       2001          11         7\n",
       "4 70083734          2000-10-05       2000          10         5\n",
       "5 50383290          1997-07-09       1997           7         9\n",
       "6 67244463          2001-11-12       2001          11        12"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "head(df3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## What's next?\n",
    "The advantage of using Spark is to work on very large data sets that would not fit into the computers memory as regular R data.frame.\n",
    "One should try to perform as much cleaning, filtering, grouping, and sorting in Spark and then pull some (`head`, `take`) or all (`collect`) rows into an R data.frame.\n",
    "See https://spark.apache.org/docs/1.6.1/api/R/index.html for a list of SparkR functions."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## What can go wrong?\n",
    "Yes, a lot of things can get wrong. The R process is 'talking' to the Spark environment, and some of the instructions and expressions that are send to Spark may cause some the background Spark processes to crash. Error messages can be very cryptic. Here are some steps:\n",
    "- Try if you're still talking to Spark by looking at your DataFrame: `head(df_deferral)`\n",
    "- If you lost the connecion to Spark you may have to reinitialize the Spark context. First kill the Spark session with `sparkR.stop()`. Then go through the initialization process\n",
    "- If nothing works restart the kernel suing the \"Kernel\" drop-down in the menu bar. Outside of the Jupyter Notebook kill your R process.\n",
    "- Check if there is a rouge process on the system by using `top` command the UNIX command line:\n",
    "<pre>\n",
    "Tasks: 986 total,   1 running, 985 sleeping,   0 stopped,   0 zombie\n",
    "%Cpu(s):  0.1 us,  0.1 sy,  0.0 ni, 99.9 id,  0.0 wa,  0.0 hi,  0.0 si,  0.0 st\n",
    "KiB Mem : 23083510+total, 83742512 free, 32968732 used, 11412386+buff/cache\n",
    "KiB Swap:  1023996 total,   520828 free,   503168 used. 19664708+avail Mem \n",
    "</pre>\n",
    "<pre>\n",
    "   PID USER      PR  NI    VIRT    RES    SHR S  %CPU %MEM     TIME+ COMMAND\n",
    "117451 molnar    20   0 12.054g 492288  25764 S   1.0  0.2   0:23.74 java\n",
    "124997 molnar    20   0  147092   3024   1424 R   1.0  0.0   0:00.17 top\n",
    "  1248 git       20   0 3345800 402508   2248 S   0.7  0.2 336:22.19 ruby\n",
    "   155 root      20   0       0      0      0 S   0.3  0.0   1:24.71 rcuos/0\n",
    "  2000 named     20   0 2496732  41816   1428 S   0.3  0.0  70:50.91 named\n",
    "  3517 git       20   0  568496 274944   4748 S   0.3  0.1   0:12.59 ruby\n",
    "100061 mgrace    20   0 17.339g 0.016t  20308 S   0.3  7.6   2:05.82 rsession\n",
    "106136 rfu3      20   0  464880  52380   7128 S   0.3  0.0   1:04.64 jupyterhub-sing\n",
    "138708 zxiao1    20   0  766980  55488   6776 S   0.3  0.0   0:38.54 python3.4\n",
    "     1 root      20   0  190936   3644   1664 S   0.0  0.0   2:46.86 systemd\n",
    "     2 root      20   0       0      0      0 S   0.0  0.0   0:00.20 kthreadd\n",
    "     3 root      20   0       0      0      0 S   0.0  0.0   0:01.25 ksoftirqd/0\n",
    "     5 root       0 -20       0      0      0 S   0.0  0.0   0:00.00 kworker/0:0H\n",
    "     8 root      rt   0       0      0      0 S   0.0  0.0   0:03.03 migration/0\n",
    "</pre>\n",
    "If there's a process on top that uses a huge amount of CPU percentage, and it's yours, kill it with `kill -9 PID` where `PID` is the process ID show in the far left column."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "### Good Bye!  --- It's good practice to close the Spark sessionm\n",
    "sparkR.stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "R",
   "language": "R",
   "name": "ir"
  },
  "language_info": {
   "codemirror_mode": "r",
   "file_extension": ".r",
   "mimetype": "text/x-r-source",
   "name": "R",
   "pygments_lexer": "r",
   "version": "3.3.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}