spark_data_pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Apache Spark Data Pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "org.apache.spark.SparkContext@11c7ba68\n",
      "org.apache.spark.sql.SparkSession@1b8f5290\n"
     ]
    }
   ],
   "source": [
    "println(sc)\n",
    "println(spark)\n",
    "import org.apache.spark.sql.DataFrame\n",
    "val sqlc = spark\n",
    "import sqlc.sql\n",
    "import sqlc.implicits._\n",
    "import org.apache.spark.ml.feature.{HashingTF, StringIndexer, VectorAssembler}\n",
    "import org.apache.spark.ml.evaluation.RegressionEvaluator\n",
    "import org.apache.spark.ml.tuning.{TrainValidationSplit, ParamGridBuilder, CrossValidator}\n",
    "// import org.apache.spark.ml.regression.LinearRegression\n",
    "import org.apache.spark.ml.regression.GBTRegressor\n",
    "// org.apache.spark.ml.classification.LogisticRegression\n",
    "import org.apache.spark.ml.Pipeline\n",
    "\n",
    "import scala.math.abs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pipeline Operators"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "implicit class RichPipes[Y](y: Y) {\n",
    "  def |>[Z](f: Y => Z) = f(y)\n",
    "  def &>[X, Z](f: (X, Y) => Z): (X => Z) = (x: X) => f(x, y)\n",
    "}\n",
    "implicit class RichFunction2[A,B,Z](f: (A,B) => Z) {\n",
    "  def %(b: B): (A => Z) = (a: A) => f(a,b)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Helper Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def table(tbl_name: String) = sql(s\"SELECT * FROM $tbl_name\")\n",
    "\n",
    "def cols_to_lower(df: DataFrame): Array[String] \n",
    "  = df.columns.map(upper => upper + \" AS \" + upper.toLowerCase)\n",
    "\n",
    "def take_to_str(x: Array[String], take_amt: Int): String \n",
    "  = x.take(take_amt).mkString(\", \")\n",
    "\n",
    "def selectStmt1(df_str: String = \"df\"): String => DataFrame \n",
    "  = (cols: String) => sql(s\"SELECT $cols FROM $df_str\")\n",
    "\n",
    "def selectStmt2(df: DataFrame, df_str: String = \"df\"): String => DataFrame \n",
    "  = (cols: String) => {\n",
    "    df.createOrReplaceTempView(df_str);\n",
    "    val result = sql(s\"SELECT $cols FROM $df_str\")\n",
    "    sql(s\"drop table $df_str\")\n",
    "    result\n",
    "  }\n",
    "\n",
    "def show(limit: Int): Unit = (df: DataFrame) => df.show(limit)\n",
    "def take(limit: Int) = (df: DataFrame) => df.take(limit)\n",
    "def grouped(size: Int) = (x: Array[String]) => x.grouped(size).toList\n",
    "\n",
    "def map_cols(f: Array[String] => String) = (x: List[Array[String]]) => x.map(f)\n",
    "def map_df(f: String => DataFrame) = (x: List[String]) => x.map(f)\n",
    "def map_show(f: DataFrame => Unit) = (x: List[DataFrame]) => x.map(f)\n",
    "\n",
    "def groupedSelects(take_amt: Int)\n",
    "  = (df: DataFrame) => { \n",
    "    df |> cols_to_lower |> grouped(take_amt) |> \n",
    "      map_cols(take_to_str _ % take_amt) |> map_df(selectStmt2(df, \"df\")) \n",
    "}\n",
    "      \n",
    "def take_cols(df: DataFrame, take_amt: Int): DataFrame = {   \n",
    "  val df_str = \"df\"; \n",
    "  val result = { \n",
    "    df.createOrReplaceTempView(df_str);\n",
    "    df |> cols_to_lower |> take_to_str _ % take_amt; \n",
    "  } |> selectStmt1(df_str)\n",
    "  sql(s\"drop table $df_str\")\n",
    "  result\n",
    "}\n",
    "\n",
    "def display(cols: Int, rows: Int) \n",
    "  = (df: DataFrame) => { df |> groupedSelects(cols) } map(_.show(rows))\n",
    "\n",
    "def createOrReplaceTempView(view_name: String) \n",
    "  = (df: DataFrame) => df.createOrReplaceTempView(view_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Read \"On-Time Performance\" Dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "'On-Time Performance'\n",
      "Record Count: 495827\n",
      "+----+-----+------------+-----------+----------+--------------+------+\n",
      "|year|month|day_of_month|day_of_week|   fl_date|unique_carrier|fl_num|\n",
      "+----+-----+------------+-----------+----------+--------------+------+\n",
      "|2016|    7|          22|          5|2016-07-22|            F9|  1462|\n",
      "|2016|    7|          22|          5|2016-07-22|            F9|  1595|\n",
      "|2016|    7|          22|          5|2016-07-22|            F9|  1596|\n",
      "|2016|    7|          22|          5|2016-07-22|            F9|  1465|\n",
      "|2016|    7|          22|          5|2016-07-22|            F9|   504|\n",
      "+----+-----+------------+-----------+----------+--------------+------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+-----------------+------+---------------+----+------------+--------+---------+\n",
      "|origin_airport_id|origin|dest_airport_id|dest|crs_dep_time|dep_time|dep_delay|\n",
      "+-----------------+------+---------------+----+------------+--------+---------+\n",
      "|            10397|   ATL|          12953| LGA|        0620|    0627|     7.00|\n",
      "|            12953|   LGA|          13303| MIA|        0930|    0934|     4.00|\n",
      "|            13303|   MIA|          12953| LGA|        1435|    1431|    -4.00|\n",
      "|            12953|   LGA|          10397| ATL|        1835|    1828|    -7.00|\n",
      "|            11292|   DEN|          12953| LGA|        1611|    1626|    15.00|\n",
      "+-----------------+------+---------------+----+------------+--------+---------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+---------+---------------+--------+----------+---------+-------+------------+\n",
      "|dep_del15|dep_delay_group|taxi_out|wheels_off|wheels_on|taxi_in|crs_arr_time|\n",
      "+---------+---------------+--------+----------+---------+-------+------------+\n",
      "|     0.00|              0|   14.00|      0641|     0835|   8.00|        0839|\n",
      "|     0.00|              0|   17.00|      0951|     1207|   5.00|        1254|\n",
      "|     0.00|             -1|   24.00|      1455|     1732|   9.00|        1733|\n",
      "|     0.00|             -1|   18.00|      1846|     2026|  12.00|        2118|\n",
      "|     1.00|              1|   21.00|      1647|     2209|   8.00|        2200|\n",
      "+---------+---------------+--------+----------+---------+-------+------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+--------+---------+---------+---------------+---------+-----------------+--------+\n",
      "|arr_time|arr_delay|arr_del15|arr_delay_group|cancelled|cancellation_code|diverted|\n",
      "+--------+---------+---------+---------------+---------+-----------------+--------+\n",
      "|    0843|     4.00|     0.00|              0|     0.00|             null|    0.00|\n",
      "|    1212|   -42.00|     0.00|             -2|     0.00|             null|    0.00|\n",
      "|    1741|     8.00|     0.00|              0|     0.00|             null|    0.00|\n",
      "|    2038|   -40.00|     0.00|             -2|     0.00|             null|    0.00|\n",
      "|    2217|    17.00|     1.00|              1|     0.00|             null|    0.00|\n",
      "+--------+---------+---------+---------------+---------+-----------------+--------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+----------------+-------------------+--------+-------+--------+--------------+-------------+\n",
      "|crs_elapsed_time|actual_elapsed_time|air_time|flights|distance|distance_group|carrier_delay|\n",
      "+----------------+-------------------+--------+-------+--------+--------------+-------------+\n",
      "|          139.00|             136.00|  114.00|   1.00|  762.00|             4|         null|\n",
      "|          204.00|             158.00|  136.00|   1.00| 1096.00|             5|         null|\n",
      "|          178.00|             190.00|  157.00|   1.00| 1096.00|             5|         null|\n",
      "|          163.00|             130.00|  100.00|   1.00|  762.00|             4|         null|\n",
      "|          229.00|             231.00|  202.00|   1.00| 1620.00|             7|         1.00|\n",
      "+----------------+-------------------+--------+-------+--------+--------------+-------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+-------------+---------+--------------+-------------------+--------------+---------------+-----------------+\n",
      "|weather_delay|nas_delay|security_delay|late_aircraft_delay|first_dep_time|total_add_gtime|longest_add_gtime|\n",
      "+-------------+---------+--------------+-------------------+--------------+---------------+-----------------+\n",
      "|         null|     null|          null|               null|          null|           null|             null|\n",
      "|         null|     null|          null|               null|          null|           null|             null|\n",
      "|         null|     null|          null|               null|          null|           null|             null|\n",
      "|         null|     null|          null|               null|          null|           null|             null|\n",
      "|         0.00|    16.00|          0.00|               0.00|          null|           null|             null|\n",
      "+-------------+---------+--------------+-------------------+--------------+---------------+-----------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+--------------------+----------------+-----------------------+-------------+------------+\n",
      "|div_airport_landings|div_reached_dest|div_actual_elapsed_time|div_arr_delay|div_distance|\n",
      "+--------------------+----------------+-----------------------+-------------+------------+\n",
      "|                   0|            null|                   null|         null|        null|\n",
      "|                   0|            null|                   null|         null|        null|\n",
      "|                   0|            null|                   null|         null|        null|\n",
      "|                   0|            null|                   null|         null|        null|\n",
      "|                   0|            null|                   null|         null|        null|\n",
      "+--------------------+----------------+-----------------------+-------------+------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "List((), (), (), (), (), (), ())"
      ]
     },
     "execution_count": 55,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val source = {\n",
    "  spark.read.format(\"csv\").option(\"header\", \"true\").load(\"data/\")\n",
    "    .createOrReplaceTempView(\"source\")\n",
    "  val col_num = table(\"source\").columns.size - 1\n",
    "  table(\"source\") |> take_cols _ % col_num\n",
    "}\n",
    "// val delay_group \n",
    "//   = spark.read.format(\"csv\").option(\"header\", \"true\").load(\"lookups/DelayGroups/\")//.orderBy(\"code\")\n",
    "// delay_group.createOrReplaceTempView(\"delay_group\")\n",
    "\n",
    "\"'On-Time Performance'\" |> println \n",
    "\"Record Count: \"+source.count |> println \n",
    "source |> display(7,5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Feature Engineering: \n",
    "##### Kept the use of UDFs to a minimum because their execution isn't optimized & I didn't want to spend \\$ on AWS EMR"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### SQL Expression String Interpolations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "// Delay Cause Indicator\n",
    "def max_of_cols(cols: Seq[String]) \n",
    "  = (a: Double, b: Double, c: Double, d: Double, e: Double) => {\n",
    "  cols.zip(List(a,b,c,d,e)).sortWith(_._2 > _._2).head._1\n",
    "  }\n",
    "val col_name_seq = Seq(\"carrier_delay\", \"weather_delay\", \"nas_delay\", \"security_delay\", \"late_aircraft_delay\")\n",
    "val max_of_cols_udf = spark.udf.register(\"max_of_cols\", max_of_cols(col_name_seq))\n",
    "\n",
    "// Weekday Indicator\n",
    "val taxi_time = \"ABS(taxi_in + taxi_out) AS taxi_time\"\n",
    "val delay_time = \"arr_delay + dep_delay AS delay_time\"\n",
    "val air_delay_time = \"actual_elapsed_time - crs_elapsed_time AS air_delay_time\"\n",
    "\n",
    "// Arrival/Departure Taxi & Delay Indicators\n",
    "def time_ind(arr: String, del: String, alias: String) \n",
    "    = s\"CASE WHEN $arr > $del THEN 'arr/in' WHEN $arr < $del THEN 'del/out' ELSE 'both' END AS $alias\"\n",
    "val taxi_ind = time_ind(\"taxi_in\",\"taxi_out\",\"taxi_ind\")\n",
    "val delay_ind = time_ind(\"arr_delay\",\"dep_delay\",\"delay_ind\")\n",
    "\n",
    "// Weekday Indicator\n",
    "def day_of_year(date_str: String) = s\"date_format($date_str, 'EEE')\"\n",
    "def weekpart(day_expr: String) \n",
    "  = s\"CASE WHEN $day_expr NOT IN ('Sat', 'Sun') THEN 'wd' ELSE 'we' END\"\n",
    "def weekday(date_str: String) = { date_str |>  day_of_year |> weekpart } + \" AS weekday\"\n",
    "\n",
    "// Cartesian Coordinate \n",
    "def cc(a: Int, z: Int) \n",
    "  = (v: String) => { \n",
    "    val i = List(a,z).map(_.toString); val A = i(0); val Z = i(1);\n",
    "    val Pi = scala.math.Pi\n",
    "    (s\"sin(2*$Pi*($v-$A)/($Z-$A))\", s\"cos(2*$Pi*($v-$A)/($Z-$A))\")\n",
    "  }\n",
    "def months_of_year_cc(date_str: String) = s\"dayofmonth($date_str)\" |> cc(0,12)\n",
    "def week_of_year_cc(date_str: String) = s\"weekofyear($date_str)\" |> cc(0,52)\n",
    "def day_of_week_cc(date_str: String) = s\"date_format($date_str, 'D') + 1\" |> cc(0,7)\n",
    "def minute_of_day_cc(hhmm: String) = {\n",
    "  s\"(INT(SUBSTRING($hhmm,1,2)) * 60) + INT(SUBSTRING($hhmm,3,2))\" |> cc(0,1440)\n",
    "}\n",
    "val (x_moy, y_moy) = \"fl_date\" |> months_of_year_cc\n",
    "val (x_woy, y_woy) = \"fl_date\" |> week_of_year_cc\n",
    "val (x_dow, y_dow) = \"fl_date\" |> day_of_week_cc\n",
    "val (x_whls_on_mod, y_whls_on_mod) = \"wheels_on\" |> minute_of_day_cc\n",
    "val (x_whls_off_mod, y_whls_off_mod) = \"wheels_off\" |> minute_of_day_cc\n",
    "\n",
    "// UNUSED: Pre-Processing for Feature Hashing Trick\n",
    "def stringify(include: Seq[String], exclude: Seq[String]) \n",
    "  =  include.diff(exclude\n",
    "           ).map(x => s\"CONCAT('$x=',$x)\"\n",
    "           ).zip(1 to include.size\n",
    "           ).map(t => t._1+\" AS str\"+t._2\n",
    "           ).mkString(\", \")\n",
    "\n",
    "// Create Bucket Case Statements\n",
    "val splits: Array[Double] = Array(Double.NegativeInfinity) ++ (\n",
    "    -15 to 180 by 15).map(_.toDouble).toSeq ++ Array(Double.PositiveInfinity)\n",
    "\n",
    "def get_bucket_cases(x: String, y: String)(splits: Array[AnyVal]) = {\n",
    "  val buckets = splits.sliding(2).toArray.map(_.mkString(\" to \"))\n",
    "  val bucket_map = 1 to buckets.size map(_ - 1) zip buckets\n",
    "  val cases = bucket_map.map(m => x+\"_bkidx = \"+m._1+\" THEN '\"+m._2+\"'\").mkString(\" WHEN \")  \n",
    "  \"CASE WHEN \"+cases+\" END AS \"+y\n",
    "  }\n",
    "\n",
    "val Seq(taxi_time_btk, air_delay_time_btk, delay_time_btk) \n",
    "  = Seq(\"taxi_time\", \"air_delay_time\", \"delay_time\"\n",
    "       ).map(x => splits |> get_bucket_cases(x,x+\"_bkt\"))\n",
    "\n",
    "// UNUSED: Creat Java SQL Timestamp\n",
    "def create_ts(date_str: String) = s\"date_format($date_str, 'yyyy-MM-dd 00:00:00')\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Interpolated SQL Expressions "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "** Potential Labels **\n",
      "ABS(taxi_in + taxi_out) AS taxi_time\n",
      "arr_delay + dep_delay AS delay_time\n",
      "\n",
      "** Time Indicators **\n",
      "CASE WHEN taxi_in > taxi_out THEN 'arr/in' WHEN taxi_in < taxi_out THEN 'del/out' ELSE 'both' END AS taxi_ind\n",
      "CASE WHEN arr_delay > dep_delay THEN 'arr/in' WHEN arr_delay < dep_delay THEN 'del/out' ELSE 'both' END AS delay_ind\n",
      "\n",
      "** Weekday **\n",
      "CASE WHEN date_format(date_str, 'EEE') NOT IN ('Sat', 'Sun') THEN 'wd' ELSE 'we' END AS weekday\n",
      "** Cartesian Coordinates **\n",
      "\n",
      "** months_of_year_cc **\n",
      "x = sin(2*3.141592653589793*(dayofmonth(fl_date)-0)/(12-0)), y = cos(2*3.141592653589793*(dayofmonth(fl_date)-0)/(12-0))\n",
      "\n",
      "** week_of_year_cc **\n",
      "x = sin(2*3.141592653589793*(weekofyear(fl_date)-0)/(52-0)), y = cos(2*3.141592653589793*(weekofyear(fl_date)-0)/(52-0))\n",
      "\n",
      "** day_of_week_cc **\n",
      "x = sin(2*3.141592653589793*(date_format(fl_date, 'D') + 1-0)/(7-0)), y = cos(2*3.141592653589793*(date_format(fl_date, 'D') + 1-0)/(7-0))\n",
      "\n",
      "** minite_of_day_cc **\n",
      "x = sin(2*3.141592653589793*((INT(SUBSTRING(wheels_on,1,2)) * 60) + INT(SUBSTRING(wheels_on,3,2))-0)/(1440-0)), y = cos(2*3.141592653589793*((INT(SUBSTRING(wheels_on,1,2)) * 60) + INT(SUBSTRING(wheels_on,3,2))-0)/(1440-0))\n",
      "x = sin(2*3.141592653589793*((INT(SUBSTRING(wheels_off,1,2)) * 60) + INT(SUBSTRING(wheels_off,3,2))-0)/(1440-0)), y = cos(2*3.141592653589793*((INT(SUBSTRING(wheels_off,1,2)) * 60) + INT(SUBSTRING(wheels_off,3,2))-0)/(1440-0))\n",
      "\n",
      "** bucket case statement **\n",
      "- taxi_time_btk\n",
      "CASE WHEN taxi_time_bkidx = 0 THEN '-Infinity to -15' WHEN taxi_time_bkidx = 1 THEN '-15 to 0' WHEN taxi_time_bkidx = 2 THEN '0 to 15' WHEN taxi_time_bkidx = 3 THEN '15 to 30' WHEN taxi_time_bkidx = 4 THEN '30 to 45' WHEN taxi_time_bkidx = 5 THEN '45 to 60' WHEN taxi_time_bkidx = 6 THEN '60 to 75' WHEN taxi_time_bkidx = 7 THEN '75 to 90' WHEN taxi_time_bkidx = 8 THEN '90 to 105' WHEN taxi_time_bkidx = 9 THEN '105 to 120' WHEN taxi_time_bkidx = 10 THEN '120 to 135' WHEN taxi_time_bkidx = 11 THEN '135 to 150' WHEN taxi_time_bkidx = 12 THEN '150 to 165' WHEN taxi_time_bkidx = 13 THEN '165 to 180' WHEN taxi_time_bkidx = 14 THEN '180 to Infinity' END AS taxi_time_bkt\n",
      "\n",
      "- air_delay_time_btk\n",
      "CASE WHEN air_delay_time_bkidx = 0 THEN '-Infinity to -15' WHEN air_delay_time_bkidx = 1 THEN '-15 to 0' WHEN air_delay_time_bkidx = 2 THEN '0 to 15' WHEN air_delay_time_bkidx = 3 THEN '15 to 30' WHEN air_delay_time_bkidx = 4 THEN '30 to 45' WHEN air_delay_time_bkidx = 5 THEN '45 to 60' WHEN air_delay_time_bkidx = 6 THEN '60 to 75' WHEN air_delay_time_bkidx = 7 THEN '75 to 90' WHEN air_delay_time_bkidx = 8 THEN '90 to 105' WHEN air_delay_time_bkidx = 9 THEN '105 to 120' WHEN air_delay_time_bkidx = 10 THEN '120 to 135' WHEN air_delay_time_bkidx = 11 THEN '135 to 150' WHEN air_delay_time_bkidx = 12 THEN '150 to 165' WHEN air_delay_time_bkidx = 13 THEN '165 to 180' WHEN air_delay_time_bkidx = 14 THEN '180 to Infinity' END AS air_delay_time_bkt\n",
      "\n",
      "- delay_time_btk\n",
      "CASE WHEN delay_time_bkidx = 0 THEN '-Infinity to -15' WHEN delay_time_bkidx = 1 THEN '-15 to 0' WHEN delay_time_bkidx = 2 THEN '0 to 15' WHEN delay_time_bkidx = 3 THEN '15 to 30' WHEN delay_time_bkidx = 4 THEN '30 to 45' WHEN delay_time_bkidx = 5 THEN '45 to 60' WHEN delay_time_bkidx = 6 THEN '60 to 75' WHEN delay_time_bkidx = 7 THEN '75 to 90' WHEN delay_time_bkidx = 8 THEN '90 to 105' WHEN delay_time_bkidx = 9 THEN '105 to 120' WHEN delay_time_bkidx = 10 THEN '120 to 135' WHEN delay_time_bkidx = 11 THEN '135 to 150' WHEN delay_time_bkidx = 12 THEN '150 to 165' WHEN delay_time_bkidx = 13 THEN '165 to 180' WHEN delay_time_bkidx = 14 THEN '180 to Infinity' END AS delay_time_bkt\n",
      "\n",
      "** Stringify **\n",
      "CONCAT('a=',a) AS str1, CONCAT('b=',b) AS str2, CONCAT('c=',c) AS str3, CONCAT('d=',d) AS str4\n"
     ]
    }
   ],
   "source": [
    "\"\\n\"+\"** Potential Labels **\"  |> println\n",
    "s\"$taxi_time\"                  |> println\n",
    "s\"$delay_time\"                 |> println\n",
    "\"\\n\"+\"** Time Indicators **\"   |> println\n",
    "s\"$taxi_ind\"                   |> println\n",
    "s\"$delay_ind\"                  |> println\n",
    "\"\\n\"+\"** Weekday **\"           |> println\n",
    "val wd = weekday(\"date_str\")\n",
    "s\"$wd\"                         |> println\n",
    "\"** Cartesian Coordinates **\"  |> println\n",
    "\"\\n\"+\"** months_of_year_cc **\" |> println\n",
    "s\"x = $x_moy\"+s\", y = $y_moy\"  |> println\n",
    "\"\\n\"+\"** week_of_year_cc **\"   |> println\n",
    "s\"x = $x_woy\"+s\", y = $y_woy\"  |> println\n",
    "\"\\n\"+\"** day_of_week_cc **\"    |> println\n",
    "s\"x = $x_dow\"+s\", y = $y_dow\"  |> println\n",
    "\"\\n\"+\"** minite_of_day_cc **\"  |> println\n",
    "s\"x = $x_whls_on_mod\"+s\", y = $y_whls_on_mod\"   |> println\n",
    "s\"x = $x_whls_off_mod\"+s\", y = $y_whls_off_mod\" |> println\n",
    "\"\\n\"+\"** bucket case statement **\"              |> println\n",
    "\"- taxi_time_btk\"                               |> println\n",
    "taxi_time_btk                                   |> println\n",
    "\"\\n\"+\"- air_delay_time_btk\"                     |> println\n",
    "air_delay_time_btk                              |> println\n",
    "\"\\n\"+\"- delay_time_btk\"                         |> println\n",
    "delay_time_btk                                  |> println\n",
    "\n",
    "\"\\n\"+\"** Stringify **\" |> println\n",
    "val (xSeq, ySeq) = (Seq(\"a\",\"b\",\"c\",\"d\",\"e\",\"f\",\"g\"), Seq(\"e\",\"f\",\"g\"))\n",
    "stringify(xSeq, ySeq)  |> println"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Transfromers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "// Bucketizer\n",
    "def bcktzr(splits: Array[Double])(x: String, y: String)\n",
    "  = (df: DataFrame) => {    \n",
    "    new Bucketizer()\n",
    "      .setInputCol(x)\n",
    "      .setOutputCol(y)\n",
    "      .setSplits(splits)\n",
    "      .transform(df) \n",
    "}\n",
    "// String Indexer\n",
    "def strIndxr(cols_for_idx: Array[String]) \n",
    "  = (df: DataFrame) => {\n",
    "    def strIndxr(col: String) \n",
    "      = new StringIndexer().setInputCol(col).setOutputCol(col+\"_idx\")\n",
    "    val strIndxr_stages = cols_for_idx.map(strIndxr)\n",
    "    val strIndxr_pipeline = new Pipeline().setStages(strIndxr_stages)\n",
    "    strIndxr_pipeline.fit(df).transform(df)\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pre-Processing & Feature Engineering"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def pre_process(df: DataFrame) = {\n",
    "  def filter_null(cols: Seq[String]) \n",
    "    = cols map{(col_str: String) => s\"$col_str IS NOT null\"} mkString(\" AND \") \n",
    "  val delay_causes = Seq(\"carrier_delay\", \"weather_delay\", \"nas_delay\", \"security_delay\", \"late_aircraft_delay\")\n",
    "  \n",
    "  val stage1 \n",
    "    = df.selectExpr(taxi_time, air_delay_time, delay_time, \"cancelled\",\n",
    "        \"DOUBLE(year)\", \"DOUBLE(month)\", \"DOUBLE(day_of_month)\", \"DOUBLE(day_of_week)\", \"fl_date\", weekday(\"fl_date\"),\n",
    "        x_moy+\" AS x_fl_date_moy\", y_moy+\" AS y_fl_date_moy\", \n",
    "        x_woy+\" AS x_fl_date_woy\", y_woy+\" AS y_fl_date_woy\", \n",
    "        x_dow+\" AS x_fl_date_dow\", y_dow+\" AS y_fl_date_dow\", \n",
    "        \"unique_carrier\", \"fl_num\", taxi_ind, delay_ind,\n",
    "        \"origin\", \"dest\", \"dep_del15\", \"arr_del15\", \"arr_delay_group\", \"dep_delay_group\", \n",
    "        x_whls_on_mod+\" AS x_whls_on_mod\", y_whls_on_mod+\" AS y_whls_on_mod\",\n",
    "        x_whls_off_mod+\" AS x_whls_off_mod\", y_whls_off_mod+\" AS y_whls_off_mod\",\n",
    "        \"flights\", \"distance\", \"distance_group\", \n",
    "        \"DOUBLE(carrier_delay)\", \"DOUBLE(weather_delay)\", \"DOUBLE(nas_delay)\", \n",
    "        \"DOUBLE(security_delay)\", \"DOUBLE(late_aircraft_delay)\"\n",
    "    ).where(delay_causes |> filter_null\n",
    "    ).selectExpr(\"*\", \n",
    "      \"max_of_cols(carrier_delay, weather_delay, nas_delay, security_delay, late_aircraft_delay) AS delay_cause\")\n",
    "    stage1.createOrReplaceTempView(\"stage1\")\n",
    "    val stage2 = { \n",
    "      stage1 |> bcktzr(splits)(\"taxi_time\", \"taxi_time_bkidx\") |> bcktzr(splits)(\"delay_time\", \"delay_time_bkidx\") |>\n",
    "         bcktzr(splits)(\"air_delay_time\", \"air_delay_time_bkidx\") \n",
    "    }.selectExpr(\"*\", taxi_time_btk, delay_time_btk, air_delay_time_btk)\n",
    "    stage2.createOrReplaceTempView(\"stage2\")\n",
    "    stage2\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "103485"
      ]
     },
     "execution_count": 60,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "{ source |> pre_process }.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+----------------+--------------------+-------------+--------------+------------------+\n",
      "|taxi_time_bkidx|delay_time_bkidx|air_delay_time_bkidx|taxi_time_bkt|delay_time_bkt|air_delay_time_bkt|\n",
      "+---------------+----------------+--------------------+-------------+--------------+------------------+\n",
      "|            3.0|             4.0|                 2.0|     15 to 30|      30 to 45|           0 to 15|\n",
      "|            3.0|             4.0|                 2.0|     15 to 30|      30 to 45|           0 to 15|\n",
      "|            9.0|             8.0|                 7.0|   105 to 120|     90 to 105|          75 to 90|\n",
      "|            5.0|            13.0|                 3.0|     45 to 60|    165 to 180|          15 to 30|\n",
      "|            4.0|            12.0|                 0.0|     30 to 45|    150 to 165|  -Infinity to -15|\n",
      "+---------------+----------------+--------------------+-------------+--------------+------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "List(())"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "{ source |> pre_process }.select(\"taxi_time_bkidx\", \"delay_time_bkidx\", \"air_delay_time_bkidx\", \n",
    "                                 \"taxi_time_bkt\", \"delay_time_bkt\", \"air_delay_time_bkt\") |> display(6,5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "103485"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pre_processed_df.count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Save tables for faster model runs by reducing Directed Acyclic Graph Complexity"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "val pre_processed = source |> pre_process\n",
    "pre_processed.write.format(\"parquet\").mode(\"overwrite\").save(\"tables/pre_process/pre_process.parquet\")\n",
    "val pre_processed_df = spark.read.format(\"parquet\").load(\"tables/pre_process/pre_process.parquet\")\n",
    "pre_processed_df.write.format(\"csv\").mode(\"overwrite\").save(\"tables/pre_process/pre_process.csv\")\n",
    "\n",
    "// Indexing Categorical Features\n",
    "val cols_for_idx = Array(\n",
    "  \"weekday\", \"unique_carrier\", \"taxi_ind\", \"delay_ind\", \"origin\", \"dest\", \n",
    "  \"dep_del15\", \"arr_del15\", \"arr_delay_group\", \"dep_delay_group\", \n",
    "  \"distance\", \"flights\", \"distance_group\"\n",
    ")\n",
    "val strIndxed = pre_processed_df |> strIndxr(cols_for_idx)\n",
    "strIndxed.write.format(\"parquet\").mode(\"overwrite\").save(\"tables/strIndxed/strIndxed.parquet\")\n",
    "val strIndxed_df = spark.read.format(\"parquet\").load(\"tables/strIndxed/strIndxed.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "103485"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "strIndxed_df.count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------+----------+---------+------+-----+------------+\n",
      "|taxi_time|air_delay_time|delay_time|cancelled|  year|month|day_of_month|\n",
      "+---------+--------------+----------+---------+------+-----+------------+\n",
      "|     29.0|           2.0|      32.0|     0.00|2016.0|  7.0|        22.0|\n",
      "|     23.0|           1.0|      31.0|     0.00|2016.0|  7.0|        22.0|\n",
      "|    118.0|          79.0|      97.0|     0.00|2016.0|  7.0|        23.0|\n",
      "|     46.0|          28.0|     168.0|     0.00|2016.0|  7.0|        23.0|\n",
      "|     32.0|         -33.0|     155.0|     0.00|2016.0|  7.0|        23.0|\n",
      "+---------+--------------+----------+---------+------+-----+------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+-----------+----------+-------+-------------------+------------------+--------------------+-------------------+\n",
      "|day_of_week|   fl_date|weekday|      x_fl_date_moy|     y_fl_date_moy|       x_fl_date_woy|      y_fl_date_woy|\n",
      "+-----------+----------+-------+-------------------+------------------+--------------------+-------------------+\n",
      "|        5.0|2016-07-22|     wd|-0.8660254037844392|0.4999999999999991|-0.35460488704253523|-0.9350162426854149|\n",
      "|        5.0|2016-07-22|     wd|-0.8660254037844392|0.4999999999999991|-0.35460488704253523|-0.9350162426854149|\n",
      "|        6.0|2016-07-23|     we|-0.5000000000000014|0.8660254037844378|-0.35460488704253523|-0.9350162426854149|\n",
      "|        6.0|2016-07-23|     we|-0.5000000000000014|0.8660254037844378|-0.35460488704253523|-0.9350162426854149|\n",
      "|        6.0|2016-07-23|     we|-0.5000000000000014|0.8660254037844378|-0.35460488704253523|-0.9350162426854149|\n",
      "+-----------+----------+-------+-------------------+------------------+--------------------+-------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+-------------------+-------------------+--------------+------+--------+---------+------+\n",
      "|      x_fl_date_dow|      y_fl_date_dow|unique_carrier|fl_num|taxi_ind|delay_ind|origin|\n",
      "+-------------------+-------------------+--------------+------+--------+---------+------+\n",
      "| 0.9749279121818268|-0.2225209339563005|            F9|   504|  arr/in|   arr/in|   DEN|\n",
      "| 0.9749279121818268|-0.2225209339563005|            F9|   509|  arr/in|   arr/in|   LGA|\n",
      "|0.43388373911757744|-0.9009688679024098|            F9|   509| del/out|  del/out|   LGA|\n",
      "|0.43388373911757744|-0.9009688679024098|            F9|  1596| del/out|   arr/in|   MIA|\n",
      "|0.43388373911757744|-0.9009688679024098|            F9|  1465|  arr/in|  del/out|   LGA|\n",
      "+-------------------+-------------------+--------------+------+--------+---------+------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+----+---------+---------+---------------+---------------+-------------------+-------------------+\n",
      "|dest|dep_del15|arr_del15|arr_delay_group|dep_delay_group|      x_whls_on_mod|      y_whls_on_mod|\n",
      "+----+---------+---------+---------------+---------------+-------------------+-------------------+\n",
      "| LGA|     1.00|     1.00|              1|              1|-0.4656145203251117| 0.8849876374630418|\n",
      "| DEN|     1.00|     1.00|              1|              1|0.25881904510252074| 0.9659258262890683|\n",
      "| DEN|     0.00|     1.00|              5|              0|0.17364817766693033|  0.984807753012208|\n",
      "| LGA|     1.00|     1.00|              6|              4|-0.9647873238288132|0.26303121445797406|\n",
      "| ATL|     1.00|     1.00|              4|              6|-0.4578739151169575| 0.8890171414857361|\n",
      "+----+---------+---------+---------------+---------------+-------------------+-------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+--------------------+-------------------+-------+--------+--------------+-------------+-------------+\n",
      "|      x_whls_off_mod|     y_whls_off_mod|flights|distance|distance_group|carrier_delay|weather_delay|\n",
      "+--------------------+-------------------+-------+--------+--------------+-------------+-------------+\n",
      "|  -0.949699126201877|-0.3131638064837495|   1.00| 1620.00|             7|          1.0|          0.0|\n",
      "|-0.19936793441719736| 0.9799247046208296|   1.00| 1620.00|             7|          0.0|          0.0|\n",
      "|-0.21217767215644653| 0.9772311064626789|   1.00| 1620.00|             7|          9.0|          0.0|\n",
      "| -0.9081431738250814| -0.418659737537428|   1.00| 1096.00|             5|         70.0|          0.0|\n",
      "|  -0.785316930880745|  0.619093949309834|   1.00|  762.00|             4|          0.0|          0.0|\n",
      "+--------------------+-------------------+-------+--------+--------------+-------------+-------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+---------+--------------+-------------------+-------------------+---------------+----------------+--------------------+\n",
      "|nas_delay|security_delay|late_aircraft_delay|        delay_cause|taxi_time_bkidx|delay_time_bkidx|air_delay_time_bkidx|\n",
      "+---------+--------------+-------------------+-------------------+---------------+----------------+--------------------+\n",
      "|     16.0|           0.0|                0.0|          nas_delay|            3.0|             4.0|                 2.0|\n",
      "|     15.0|           0.0|                1.0|          nas_delay|            3.0|             4.0|                 2.0|\n",
      "|     79.0|           0.0|                0.0|          nas_delay|            9.0|             8.0|                 7.0|\n",
      "|     28.0|           0.0|                0.0|      carrier_delay|            5.0|            13.0|                 3.0|\n",
      "|      0.0|           0.0|               61.0|late_aircraft_delay|            4.0|            12.0|                 0.0|\n",
      "+---------+--------------+-------------------+-------------------+---------------+----------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+-------------+--------------+------------------+-----------+------------------+------------+-------------+\n",
      "|taxi_time_bkt|delay_time_bkt|air_delay_time_bkt|weekday_idx|unique_carrier_idx|taxi_ind_idx|delay_ind_idx|\n",
      "+-------------+--------------+------------------+-----------+------------------+------------+-------------+\n",
      "|     15 to 30|      30 to 45|           0 to 15|        0.0|               9.0|         0.0|          1.0|\n",
      "|     15 to 30|      30 to 45|           0 to 15|        0.0|               9.0|         0.0|          1.0|\n",
      "|   105 to 120|     90 to 105|          75 to 90|        1.0|               9.0|         1.0|          0.0|\n",
      "|     45 to 60|    165 to 180|          15 to 30|        1.0|               9.0|         1.0|          1.0|\n",
      "|     30 to 45|    150 to 165|  -Infinity to -15|        1.0|               9.0|         0.0|          0.0|\n",
      "+-------------+--------------+------------------+-----------+------------------+------------+-------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+----------+--------+-------------+-------------+-------------------+-------------------+------------+\n",
      "|origin_idx|dest_idx|dep_del15_idx|arr_del15_idx|arr_delay_group_idx|dep_delay_group_idx|distance_idx|\n",
      "+----------+--------+-------------+-------------+-------------------+-------------------+------------+\n",
      "|      17.0|     0.0|          0.0|          0.0|                0.0|                1.0|        18.0|\n",
      "|       1.0|    22.0|          0.0|          0.0|                0.0|                1.0|        18.0|\n",
      "|       1.0|    22.0|          1.0|          0.0|                5.0|                2.0|        18.0|\n",
      "|       6.0|     0.0|          0.0|          0.0|                6.0|                5.0|         6.0|\n",
      "|       1.0|     7.0|          0.0|          0.0|                3.0|                8.0|         4.0|\n",
      "+----------+--------+-------------+-------------+-------------------+-------------------+------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+-----------+------------------+\n",
      "|flights_idx|distance_group_idx|\n",
      "+-----------+------------------+\n",
      "|        0.0|               7.0|\n",
      "|        0.0|               7.0|\n",
      "|        0.0|               7.0|\n",
      "|        0.0|               0.0|\n",
      "|        0.0|               2.0|\n",
      "+-----------+------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "List((), (), (), (), (), (), (), (), ())"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "strIndxed_df |> display(7,5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Feature Counts For GBTRegressor MaxBin tuning"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(year_count,2)\n",
      "(month_count,12)\n",
      "(day_of_month_count,31)\n",
      "(day_of_week_count,7)\n",
      "(x_fl_date_moy_count,28)\n",
      "(y_fl_date_moy_count,28)\n",
      "(x_fl_date_woy_count,50)\n",
      "(y_fl_date_woy_count,50)\n",
      "(x_fl_date_dow_count,326)\n",
      "(y_fl_date_dow_count,231)\n",
      "(weekday_idx_count,2)\n",
      "(unique_carrier_idx_count,13)\n",
      "(taxi_ind_idx_count,3)\n",
      "(delay_ind_idx_count,3)\n",
      "(origin_idx_count,106)\n",
      "(dest_idx_count,105)\n",
      "(dep_del15_idx_count,2)\n",
      "(arr_del15_idx_count,1)\n",
      "(arr_delay_group_idx_count,12)\n",
      "(dep_delay_group_idx_count,15)\n",
      "(distance_idx_count,191)\n",
      "(flights_idx_count,1)\n",
      "(distance_group_idx_count,11)\n",
      "(taxi_time_bkt_count,13)\n",
      "(delay_time_bkt_count,14)\n",
      "(air_delay_time_bkt_count,15)\n",
      "(carrier_delay_count,699)\n",
      "(weather_delay_count,382)\n",
      "(nas_delay_count,423)\n",
      "(security_delay_count,101)\n",
      "(late_aircraft_delay_count,466)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "Array((), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), ())"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "strIndxed_df |> createOrReplaceTempView(\"test\")\n",
    "val idxed_cols = cols_for_idx.map(x => x+\"_idx\")\n",
    "val feature_cols = Array(\"year\", \"month\", \"day_of_month\", \"day_of_week\",\n",
    "    \"x_fl_date_moy\", \"y_fl_date_moy\", \"x_fl_date_woy\", \"y_fl_date_woy\", \"x_fl_date_dow\", \"y_fl_date_dow\"\n",
    "    ) ++ idxed_cols ++ Array(\"taxi_time_bkt\", \"delay_time_bkt\", \"air_delay_time_bkt\", \n",
    "                             \"carrier_delay\", \"weather_delay\", \"nas_delay\", \"security_delay\", \"late_aircraft_delay\")\n",
    "val count_cols = feature_cols.map(x => s\"COUNT(DISTINCT($x)) AS $x\"+\"_count\").mkString(\", \")\n",
    "val df = sql(s\"SELECT $count_cols FROM test\")\n",
    "df.columns zip df.collect()(0).toSeq map println"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Cross-Validated Gradient Boosted Tree Regressor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array(weekday_idx, unique_carrier_idx, taxi_ind_idx, delay_ind_idx, origin_idx, dest_idx, dep_del15_idx, arr_del15_idx, arr_delay_group_idx, dep_delay_group_idx, distance_idx, flights_idx, distance_group_idx)"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cols_for_idx.map(x => x+\"_idx\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "103485\n"
     ]
    }
   ],
   "source": [
    "def cv_gbt(df: DataFrame) = {\n",
    "  val idxed_cols = cols_for_idx.map(x => x+\"_idx\")\n",
    "  val feature_cols = Array(\"year\", \"month\", \"day_of_month\", \"day_of_week\",\n",
    "    \"x_fl_date_moy\", \"y_fl_date_moy\", \"x_fl_date_woy\", \"y_fl_date_woy\", \"x_fl_date_dow\", \"y_fl_date_dow\"\n",
    "    ) ++ idxed_cols ++ Array(\"x_whls_on_mod\", \"y_whls_on_mod\", \"x_whls_off_mod\", \"y_whls_off_mod\",\n",
    "    \"carrier_delay\", \"weather_delay\", \"nas_delay\", \"security_delay\", \"late_aircraft_delay\")\n",
    "    \n",
    "  val featAssembler = new VectorAssembler(\n",
    "    ).setInputCols(feature_cols\n",
    "    ).setOutputCol(\"feat_vec\")\n",
    "    \n",
    "  var df_wFeatVec = featAssembler.transform(df).drop(\n",
    "    \"weekday_idx\", \"unique_carrier_idx\", \"taxi_ind_idx\", \"delay_ind_idx\", \n",
    "    \"origin_idx\", \"dest_idx\", \"dep_del15_idx\", \"arr_del15_idx\", \"arr_delay_group_idx\", \"dep_delay_group_idx\", \n",
    "    \"distance_idx\", \"flights_idx\", \"distance_group_idx\")\n",
    "    \n",
    "  println(df_wFeatVec.count())\n",
    "    \n",
    "  val Array(training, test) \n",
    "    = df_wFeatVec.randomSplit(Array(0.8, 0.2), seed = 12345)\n",
    "        .map(_.withColumnRenamed(\"taxi_time\", \"label\").withColumnRenamed(\"feat_vec\", \"features\"))\n",
    "    \n",
    "  val gbt = new GBTRegressor()\n",
    "\n",
    "  val paramGrid = new ParamGridBuilder() // Expensive\n",
    "    .addGrid(gbt.labelCol, Array(\"label\"))\n",
    "    .addGrid(gbt.featuresCol, Array(\"features\"))\n",
    "    .addGrid(gbt.maxIter, Array(10))\n",
    "    .addGrid(gbt.maxBins, Array(250))\n",
    "    .build()\n",
    "    \n",
    "  val cv = new CrossValidator() // Expensive\n",
    "    .setEstimator(gbt)\n",
    "    .setEvaluator(new RegressionEvaluator)\n",
    "    .setEstimatorParamMaps(paramGrid)\n",
    "    .setNumFolds(3) \n",
    "\n",
    "  val cvModel = cv.fit(training)\n",
    "  // TODO: Select more for EDA\n",
    "  cvModel.transform(test).drop(\"features\") //.select(\"features\", \"label\", \"prediction\")\n",
    "}\n",
    "{ strIndxed_df |> cv_gbt }.write.format(\"parquet\").mode(\"overwrite\").save(\"tables/gbt/gbt.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Save For Exploratory Data Analysis in a Python notebook\n",
    "##### Note: Although the ipynb can run Apache Toree's PySpark kernel as well as Spark's (Scala), there was a matplotlib dependency issue I didnt wan't to solve at the time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Name: Syntax Error.\n",
       "Message: \n",
       "StackTrace: "
      ]
     },
     "execution_count": 67,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val gbt_res_df = spark.read.format(\"parquet\").load(\"tables/gbt/gbt.parquet/\")\n",
    "// gbt_res_df.write.format(\"csv\").mode(\"overwrite\").save(\"tables/gbt/gbt.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "20870"
      ]
     },
     "execution_count": 68,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "gbt_res_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "20870"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "gbt_res_df.count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Regression Metrics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Root Mean Squared Error (rmse) - 12.060027132075415\n",
      "      Mean Squared Error (mse) - 145.44425442639516\n",
      "   Median Absolute Error (mae) - 8.552334236413296\n",
      "Coef of Determination     (r2) - 0.6533716108894303\n"
     ]
    }
   ],
   "source": [
    "val reg_eval = new RegressionEvaluator(\n",
    "  ).setLabelCol(\"label\"\n",
    "  ).setPredictionCol(\"prediction\")\n",
    "\n",
    "val List(rmse, mse, r2, mae) \n",
    "  = List(\"rmse\", \"mse\", \"r2\",\"mae\"\n",
    "      ).map(reg_eval.setMetricName(_).evaluate(gbt_res_df))\n",
    "\n",
    "s\"Root Mean Squared Error (rmse) - $rmse\" |> println\n",
    "s\"      Mean Squared Error (mse) - $mse\"  |> println\n",
    "s\"   Median Absolute Error (mae) - $mae\"  |> println\n",
    "s\"Coef of Determination     (r2) - $r2\"   |> println"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Backlog"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "// UNUSED: Hashing Trick\n",
    "def hashing_trick(numFeatures: Int, binary: Boolean) \n",
    "    = (df: DataFrame) => {\n",
    "      val include = df.drop(\"taxi_time\", \"delay_time\", \"cancelled\").columns\n",
    "      val exclude = df.select(\"taxi_time\", \"delay_time\", \"cancelled\").columns\n",
    "      val stringed_exprs = stringify(include, exclude)\n",
    "      val stringed_cols = (1 to include.size).map(\"str\"+_.toString).mkString(\", \")\n",
    "      val stringed_array = \"ARRAY(\"+stringed_cols+\") AS raw_feats\"\n",
    "      df.createOrReplaceTempView(\"df\")\n",
    "      val all_cols_df = sql(s\"SELECT *, $stringed_exprs FROM df\").selectExpr(\"*\", stringed_array)\n",
    "      all_cols_df.createOrReplaceTempView(\"all_cols_df\")\n",
    "      val selected_cols = all_cols_df.columns.take(34) ++ all_cols_df.columns.takeRight(1) mkString(\", \")\n",
    "      val df2 = sql(s\"SELECT $selected_cols FROM all_cols_df\")\n",
    "\n",
    "      val hashingTF = new HashingTF()\n",
    "       .setInputCol(\"raw_feats\")\n",
    "       .setOutputCol(\"hashed_feats\")\n",
    "       .setNumFeatures(include.size)\n",
    "       .setBinary(binary)\n",
    "\n",
    "      hashingTF.transform(df2)\n",
    "  }"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Apache Toree - Scala",
   "language": "scala",
   "name": "apache_toree_scala"
  },
  "language_info": {
   "file_extension": ".scala",
   "name": "scala",
   "version": "2.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}