26 Introduction to PySpark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Getting to know PySpark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "airports.csv, flights_skalp.csv, planes.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:24:40.220166Z",
     "start_time": "2019-02-16T11:24:39.703519Z"
    }
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "\n",
    "%matplotlib inline\n",
    "\n",
    "import pyspark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-01-02T20:12:37.147855Z",
     "start_time": "2019-01-02T20:12:37.126828Z"
    }
   },
   "source": [
    "## What is Spark, anyway?\n",
    "\n",
    "Spark is a platform for cluster computing. Spark lets you spread data and computations over clusters with multiple nodes (think of each node as a separate computer). Splitting up your data makes it easier to work with very large datasets because each node only works with a small amount of data.\n",
    "\n",
    "As each node works on its own subset of the total data, it also carries out a part of the total calculations required, so that both data processing and computation are performed in parallel over the nodes in the cluster. It is a fact that parallel computation can make certain types of programming tasks much faster.\n",
    "\n",
    "However, with greater computing power comes greater complexity.\n",
    "\n",
    "Deciding whether or not Spark is the best solution for your problem takes some experience, but you can consider questions like:\n",
    "\n",
    "    Is my data too big to work with on a single machine?\n",
    "    Can my calculations be easily parallelized?\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using Spark in Python\n",
    "\n",
    "The first step in using Spark is connecting to a cluster.\n",
    "\n",
    "In practice, the cluster will be hosted on a remote machine that's connected to all other nodes. There will be one computer, called the master that manages splitting up the data and the computations. The master is connected to the rest of the computers in the cluster, which are called slaves. The master sends the slaves data and calculations to run, and they send their results back to the master.\n",
    "\n",
    "When you're just getting started with Spark it's simpler to just run a cluster locally. Thus, for this course, instead of connecting to another computer, all computations will be run on DataCamp's servers in a simulated cluster.\n",
    "\n",
    "Creating the connection is as simple as creating an instance of the SparkContext class. The class constructor takes a few optional arguments that allow you to specify the attributes of the cluster you're connecting to.\n",
    "\n",
    "An object holding all these attributes can be created with the SparkConf() constructor. Take a look at the documentation for all the details!\n",
    "\n",
    "For the rest of this course you'll have a SparkContext called sc already available in your workspace."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Examining The SparkContext\n",
    "\n",
    "In this exercise you'll get familiar with the SparkContext.\n",
    "\n",
    "You'll probably notice that code takes longer to run than you might expect. This is because Spark is some serious software. It takes more time to start up than you might be used to. You may also find that running simpler computations might take longer than expected. That's because all the optimizations that Spark has under its hood are designed for complicated operations with big data sets. That means that for simple or small problems Spark may actually perform worse than some other solutions!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:25:08.038909Z",
     "start_time": "2019-02-16T11:25:08.033924Z"
    }
   },
   "outputs": [],
   "source": [
    "# Import SparkSession from pyspark.sql\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:25:09.821309Z",
     "start_time": "2019-02-16T11:25:09.735543Z"
    }
   },
   "outputs": [],
   "source": [
    "# Create my_spark\n",
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:25:28.531805Z",
     "start_time": "2019-02-16T11:25:28.525837Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<pyspark.sql.session.SparkSession object at 0x00000153C53342E8>\n",
      "2.4.0\n"
     ]
    }
   ],
   "source": [
    "# Verify SparkContext\n",
    "print(spark)\n",
    "\n",
    "# Print Spark version\n",
    "print(spark.version)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using DataFrames\n",
    "\n",
    "Spark's core data structure is the Resilient Distributed Dataset (RDD). This is a low level object that lets Spark work its magic by splitting data across multiple nodes in the cluster. However, RDDs are hard to work with directly, so in this course you'll be using the Spark DataFrame abstraction built on top of RDDs.\n",
    "\n",
    "The Spark DataFrame was designed to behave a lot like a SQL table (a table with variables in the columns and observations in the rows). Not only are they easier to understand, DataFrames are also more optimized for complicated operations than RDDs.\n",
    "\n",
    "When you start modifying and combining columns and rows of data, there are many ways to arrive at the same result, but some often take much longer than others. When using RDDs, it's up to the data scientist to figure out the right way to optimize the query, but the DataFrame implementation has much of this optimization built in!\n",
    "\n",
    "To start working with Spark DataFrames, you first have to create a SparkSession object from your SparkContext. You can think of the SparkContext as your connection to the cluster and the SparkSession as your interface with that connection.\n",
    "\n",
    "Remember, for the rest of this course you'll have a SparkSession called spark available in your workspace!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating a SparkSession\n",
    "\n",
    "We've already created a SparkSession for you called spark, but what if you're not sure there already is one? Creating multiple SparkSessions and SparkContexts can cause issues, so it's best practice to use the SparkSession.builder.getOrCreate() method. This returns an existing SparkSession if there's already one in the environment, or creates a new one if necessary!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:25:33.521058Z",
     "start_time": "2019-02-16T11:25:33.517066Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<pyspark.sql.session.SparkSession object at 0x00000153C53342E8>\n"
     ]
    }
   ],
   "source": [
    "# Import SparkSession from pyspark.sql\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# Create my_spark\n",
    "spark = SparkSession.builder.getOrCreate()\n",
    "\n",
    "# Print my_spark\n",
    "print(spark)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Viewing tables\n",
    "\n",
    "Once you've created a SparkSession, you can start poking around to see what data is in your cluster!\n",
    "\n",
    "Your SparkSession has an attribute called catalog which lists all the data inside the cluster. This attribute has a few methods for extracting different pieces of information.\n",
    "\n",
    "One of the most useful is the .listTables() method, which returns the names of all the tables in your cluster as a list."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:25:40.084929Z",
     "start_time": "2019-02-16T11:25:36.008808Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[]\n"
     ]
    }
   ],
   "source": [
    "# Print the tables in the catalog\n",
    "print(spark.catalog.listTables())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Are you query-ious?\n",
    "\n",
    "One of the advantages of the DataFrame interface is that you can run SQL queries on the tables in your Spark cluster. If you don't have any experience with SQL, don't worry (you can take our Introduction to SQL course!), we'll provide you with queries!\n",
    "\n",
    "As you saw in the last exercise, one of the tables in your cluster is the flights table. This table contains a row for every flight that left Portland International Airport (PDX) or Seattle-Tacoma International Airport (SEA) in 2014 and 2015.\n",
    "\n",
    "Running a query on this table is as easy as using the .sql() method on your SparkSession. This method takes a string containing the query and returns a DataFrame with the results!\n",
    "\n",
    "If you look closely, you'll notice that the table flights is only mentioned in the query, not as an argument to any of the methods. This is because there isn't a local object in your environment that holds that data, so it wouldn't make sense to pass the table as an argument.\n",
    "\n",
    "Remember, we've already created a SparkSession called spark in your workspace."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:25:42.430623Z",
     "start_time": "2019-02-16T11:25:42.060610Z"
    }
   },
   "outputs": [
    {
     "ename": "AnalysisException",
     "evalue": "'Table or view not found: flights; line 1 pos 5'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     62\u001b[0m         \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 63\u001b[1;33m             \u001b[1;32mreturn\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m*\u001b[0m\u001b[0ma\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     64\u001b[0m         \u001b[1;32mexcept\u001b[0m \u001b[0mpy4j\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mprotocol\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mPy4JJavaError\u001b[0m \u001b[1;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\py4j\\protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m    327\u001b[0m                     \u001b[1;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 328\u001b[1;33m                     format(target_id, \".\", name), value)\n\u001b[0m\u001b[0;32m    329\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling o21.sql.\n: org.apache.spark.sql.AnalysisException: Table or view not found: flights; line 1 pos 5\r\n\tat org.apache.spark.sql.catalyst.analysis.package$AnalysisErrorAt.failAnalysis(package.scala:47)\r\n\tat org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveRelations$.org$apache$spark$sql$catalyst$analysis$Analyzer$ResolveRelations$$lookupTableFromCatalog(Analyzer.scala:733)\r\n\tat org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveRelations$.resolveRelation(Analyzer.scala:685)\r\n\tat org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveRelations$$anonfun$apply$8.applyOrElse(Analyzer.scala:715)\r\n\tat org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveRelations$$anonfun$apply$8.applyOrElse(Analyzer.scala:708)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1$$anonfun$apply$1.apply(AnalysisHelper.scala:90)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1$$anonfun$apply$1.apply(AnalysisHelper.scala:90)\r\n\tat org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1.apply(AnalysisHelper.scala:89)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1.apply(AnalysisHelper.scala:86)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$.allowInvokingTransformsInAnalyzer(AnalysisHelper.scala:194)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$class.resolveOperatorsUp(AnalysisHelper.scala:86)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.resolveOperatorsUp(LogicalPlan.scala:29)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1$$anonfun$1.apply(AnalysisHelper.scala:87)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1$$anonfun$1.apply(AnalysisHelper.scala:87)\r\n\tat org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:326)\r\n\tat org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)\r\n\tat org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:324)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1.apply(AnalysisHelper.scala:87)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1.apply(AnalysisHelper.scala:86)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$.allowInvokingTransformsInAnalyzer(AnalysisHelper.scala:194)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$class.resolveOperatorsUp(AnalysisHelper.scala:86)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.resolveOperatorsUp(LogicalPlan.scala:29)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1$$anonfun$1.apply(AnalysisHelper.scala:87)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1$$anonfun$1.apply(AnalysisHelper.scala:87)\r\n\tat org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:326)\r\n\tat org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)\r\n\tat org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:324)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1.apply(AnalysisHelper.scala:87)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1.apply(AnalysisHelper.scala:86)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$.allowInvokingTransformsInAnalyzer(AnalysisHelper.scala:194)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$class.resolveOperatorsUp(AnalysisHelper.scala:86)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.resolveOperatorsUp(LogicalPlan.scala:29)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1$$anonfun$1.apply(AnalysisHelper.scala:87)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1$$anonfun$1.apply(AnalysisHelper.scala:87)\r\n\tat org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:326)\r\n\tat org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)\r\n\tat org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:324)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1.apply(AnalysisHelper.scala:87)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$$anonfun$resolveOperatorsUp$1.apply(AnalysisHelper.scala:86)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$.allowInvokingTransformsInAnalyzer(AnalysisHelper.scala:194)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$class.resolveOperatorsUp(AnalysisHelper.scala:86)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.resolveOperatorsUp(LogicalPlan.scala:29)\r\n\tat org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveRelations$.apply(Analyzer.scala:708)\r\n\tat org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveRelations$.apply(Analyzer.scala:654)\r\n\tat org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1$$anonfun$apply$1.apply(RuleExecutor.scala:87)\r\n\tat org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1$$anonfun$apply$1.apply(RuleExecutor.scala:84)\r\n\tat scala.collection.LinearSeqOptimized$class.foldLeft(LinearSeqOptimized.scala:124)\r\n\tat scala.collection.immutable.List.foldLeft(List.scala:84)\r\n\tat org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1.apply(RuleExecutor.scala:84)\r\n\tat org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1.apply(RuleExecutor.scala:76)\r\n\tat scala.collection.immutable.List.foreach(List.scala:392)\r\n\tat org.apache.spark.sql.catalyst.rules.RuleExecutor.execute(RuleExecutor.scala:76)\r\n\tat org.apache.spark.sql.catalyst.analysis.Analyzer.org$apache$spark$sql$catalyst$analysis$Analyzer$$executeSameContext(Analyzer.scala:127)\r\n\tat org.apache.spark.sql.catalyst.analysis.Analyzer.execute(Analyzer.scala:121)\r\n\tat org.apache.spark.sql.catalyst.analysis.Analyzer$$anonfun$executeAndCheck$1.apply(Analyzer.scala:106)\r\n\tat org.apache.spark.sql.catalyst.analysis.Analyzer$$anonfun$executeAndCheck$1.apply(Analyzer.scala:105)\r\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$.markInAnalyzer(AnalysisHelper.scala:201)\r\n\tat org.apache.spark.sql.catalyst.analysis.Analyzer.executeAndCheck(Analyzer.scala:105)\r\n\tat org.apache.spark.sql.execution.QueryExecution.analyzed$lzycompute(QueryExecution.scala:57)\r\n\tat org.apache.spark.sql.execution.QueryExecution.analyzed(QueryExecution.scala:55)\r\n\tat org.apache.spark.sql.execution.QueryExecution.assertAnalyzed(QueryExecution.scala:47)\r\n\tat org.apache.spark.sql.Dataset$.ofRows(Dataset.scala:79)\r\n\tat org.apache.spark.sql.SparkSession.sql(SparkSession.scala:642)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\r\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\r\n\tat java.lang.reflect.Method.invoke(Method.java:498)\r\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\r\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\r\n\tat py4j.Gateway.invoke(Gateway.java:282)\r\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\r\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\r\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\r\n\tat java.lang.Thread.run(Thread.java:748)\r\nCaused by: org.apache.spark.sql.catalyst.analysis.NoSuchTableException: Table or view 'flights' not found in database 'default';\r\n\tat org.apache.spark.sql.catalyst.catalog.ExternalCatalog$class.requireTableExists(ExternalCatalog.scala:48)\r\n\tat org.apache.spark.sql.catalyst.catalog.InMemoryCatalog.requireTableExists(InMemoryCatalog.scala:45)\r\n\tat org.apache.spark.sql.catalyst.catalog.InMemoryCatalog.getTable(InMemoryCatalog.scala:326)\r\n\tat org.apache.spark.sql.catalyst.catalog.ExternalCatalogWithListener.getTable(ExternalCatalogWithListener.scala:138)\r\n\tat org.apache.spark.sql.catalyst.catalog.SessionCatalog.lookupRelation(SessionCatalog.scala:701)\r\n\tat org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveRelations$.org$apache$spark$sql$catalyst$analysis$Analyzer$ResolveRelations$$lookupTableFromCatalog(Analyzer.scala:730)\r\n\t... 73 more\r\n",
      "\nDuring handling of the above exception, another exception occurred:\n",
      "\u001b[1;31mAnalysisException\u001b[0m                         Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-9-dd61526e917e>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m\u001b[0m\n\u001b[0;32m      3\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      4\u001b[0m \u001b[1;31m# Get the first 10 rows of flights\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 5\u001b[1;33m \u001b[0mflights10\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mspark\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msql\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mquery\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      6\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      7\u001b[0m \u001b[1;31m# Show the results\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\session.py\u001b[0m in \u001b[0;36msql\u001b[1;34m(self, sqlQuery)\u001b[0m\n\u001b[0;32m    765\u001b[0m         \u001b[1;33m[\u001b[0m\u001b[0mRow\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mf1\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mf2\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;34mu'row1'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mRow\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mf1\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m2\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mf2\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;34mu'row2'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mRow\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mf1\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m3\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mf2\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;34mu'row3'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    766\u001b[0m         \"\"\"\n\u001b[1;32m--> 767\u001b[1;33m         \u001b[1;32mreturn\u001b[0m \u001b[0mDataFrame\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jsparkSession\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msql\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0msqlQuery\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_wrapped\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    768\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    769\u001b[0m     \u001b[1;33m@\u001b[0m\u001b[0msince\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;36m2.0\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\py4j\\java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[1;34m(self, *args)\u001b[0m\n\u001b[0;32m   1255\u001b[0m         \u001b[0manswer\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1256\u001b[0m         return_value = get_return_value(\n\u001b[1;32m-> 1257\u001b[1;33m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[0;32m   1258\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1259\u001b[0m         \u001b[1;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     67\u001b[0m                                              e.java_exception.getStackTrace()))\n\u001b[0;32m     68\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'org.apache.spark.sql.AnalysisException: '\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 69\u001b[1;33m                 \u001b[1;32mraise\u001b[0m \u001b[0mAnalysisException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m': '\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     70\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'org.apache.spark.sql.catalyst.analysis'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     71\u001b[0m                 \u001b[1;32mraise\u001b[0m \u001b[0mAnalysisException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m': '\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mAnalysisException\u001b[0m: 'Table or view not found: flights; line 1 pos 5'"
     ]
    }
   ],
   "source": [
    "# Don't change this query\n",
    "query = \"FROM flights SELECT * LIMIT 10\"\n",
    "\n",
    "# Get the first 10 rows of flights\n",
    "flights10 = spark.sql(query)\n",
    "\n",
    "# Show the results\n",
    "flights10.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pandafy a Spark DataFrame\n",
    "\n",
    "Suppose you've run a query on your huge dataset and aggregated it down to something a little more manageable.\n",
    "\n",
    "Sometimes it makes sense to then take that table and work with it locally using a tool like pandas. Spark DataFrames make that easy with the .toPandas() method. Calling this method on a Spark DataFrame returns the corresponding pandas DataFrame. It's as simple as that!\n",
    "\n",
    "This time the query counts the number of flights to each airport from SEA and PDX.\n",
    "\n",
    "Remember, there's already a SparkSession called spark in your workspace!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Don't change this query\n",
    "query = \"SELECT origin, dest, COUNT(*) as N FROM flights GROUP BY origin, dest\"\n",
    "\n",
    "# Run the query\n",
    "flight_counts = spark.sql(query)\n",
    "\n",
    "# Convert the results to a pandas DataFrame\n",
    "pd_counts = flight_counts.toPandas()\n",
    "\n",
    "# Print the head of pd_counts\n",
    "print(pd_counts.head())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Put some Spark in your data\n",
    "\n",
    "In the last exercise, you saw how to move data from Spark to pandas. However, maybe you want to go the other direction, and put a pandas DataFrame into a Spark cluster! The SparkSession class has a method for this as well.\n",
    "\n",
    "The .createDataFrame() method takes a pandas DataFrame and returns a Spark DataFrame.\n",
    "\n",
    "The output of this method is stored locally, not in the SparkSession catalog. This means that you can use all the Spark DataFrame methods on it, but you can't access the data in other contexts.\n",
    "\n",
    "For example, a SQL query (using the .sql() method) that references your DataFrame will throw an error. To access the data in this way, you have to save it as a temporary table.\n",
    "\n",
    "You can do this using the .createTempView() Spark DataFrame method, which takes as its only argument the name of the temporary table you'd like to register. This method registers the DataFrame as a table in the catalog, but as this table is temporary, it can only be accessed from the specific SparkSession used to create the Spark DataFrame.\n",
    "\n",
    "There is also the method .createOrReplaceTempView(). This safely creates a new temporary table if nothing was there before, or updates an existing table if one was already defined. You'll use this method to avoid running into problems with duplicate tables.\n",
    "\n",
    "Check out the diagram to see all the different ways your Spark data structures interact with each other."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:27:51.364340Z",
     "start_time": "2019-02-16T11:27:50.874669Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[]\n",
      "[Table(name='temp', database=None, description=None, tableType='TEMPORARY', isTemporary=True)]\n"
     ]
    }
   ],
   "source": [
    "# Create pd_temp\n",
    "pd_temp = pd.DataFrame(np.random.random(10))\n",
    "\n",
    "# Create spark_temp from pd_temp\n",
    "spark_temp = spark.createDataFrame(pd_temp)\n",
    "\n",
    "# Examine the tables in the catalog\n",
    "print(spark.catalog.listTables())\n",
    "\n",
    "# Add spark_temp to the catalog\n",
    "spark_temp.createOrReplaceTempView('temp')\n",
    "\n",
    "# Examine the tables in the catalog again\n",
    "print(spark.catalog.listTables())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dropping the middle man\n",
    "\n",
    "Now you know how to put data into Spark via pandas, but you're probably wondering why deal with pandas at all? Wouldn't it be easier to just read a text file straight into Spark? Of course it would!\n",
    "\n",
    "Luckily, your SparkSession has a .read attribute which has several methods for reading different data sources into Spark DataFrames. Using these you can create a DataFrame from a .csv file just like with regular pandas DataFrames!\n",
    "\n",
    "The variable file_path is a string with the path to the file airports.csv. This file contains information about different airports all over the world.\n",
    "\n",
    "A SparkSession named spark is available in your workspace."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:28:18.051565Z",
     "start_time": "2019-02-16T11:28:17.125971Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+----------------+-----------------+----+---+---+\n",
      "|faa|                name|             lat|              lon| alt| tz|dst|\n",
      "+---+--------------------+----------------+-----------------+----+---+---+\n",
      "|04G|   Lansdowne Airport|      41.1304722|      -80.6195833|1044| -5|  A|\n",
      "|06A|Moton Field Munic...|      32.4605722|      -85.6800278| 264| -5|  A|\n",
      "|06C| Schaumburg Regional|      41.9893408|      -88.1012428| 801| -6|  A|\n",
      "|06N|     Randall Airport|       41.431912|      -74.3915611| 523| -5|  A|\n",
      "|09J|Jekyll Island Air...|      31.0744722|      -81.4277778|  11| -4|  A|\n",
      "|0A9|Elizabethton Muni...|      36.3712222|      -82.1734167|1593| -4|  A|\n",
      "|0G6|Williams County A...|      41.4673056|      -84.5067778| 730| -5|  A|\n",
      "|0G7|Finger Lakes Regi...|      42.8835647|      -76.7812318| 492| -5|  A|\n",
      "|0P2|Shoestring Aviati...|      39.7948244|      -76.6471914|1000| -5|  U|\n",
      "|0S9|Jefferson County ...|      48.0538086|     -122.8106436| 108| -8|  A|\n",
      "|0W3|Harford County Ai...|      39.5668378|      -76.2024028| 409| -5|  A|\n",
      "|10C|  Galt Field Airport|      42.4028889|      -88.3751111| 875| -6|  U|\n",
      "|17G|Port Bucyrus-Craw...|      40.7815556|      -82.9748056|1003| -5|  A|\n",
      "|19A|Jackson County Ai...|      34.1758638|      -83.5615972| 951| -4|  U|\n",
      "|1A3|Martin Campbell F...|      35.0158056|      -84.3468333|1789| -4|  A|\n",
      "|1B9| Mansfield Municipal|      42.0001331|      -71.1967714| 122| -5|  A|\n",
      "|1C9|Frazier Lake Airpark|54.0133333333333|-124.768333333333| 152| -8|  A|\n",
      "|1CS|Clow Internationa...|      41.6959744|      -88.1292306| 670| -6|  U|\n",
      "|1G3|  Kent State Airport|      41.1513889|      -81.4151111|1134| -4|  A|\n",
      "|1OH|     Fortman Airport|      40.5553253|      -84.3866186| 885| -5|  U|\n",
      "+---+--------------------+----------------+-----------------+----+---+---+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Don't change this file path\n",
    "file_path = \"airports.csv\"\n",
    "\n",
    "# Read in the airports data\n",
    "airports = spark.read.csv(file_path, header=True)\n",
    "\n",
    "# Show the data\n",
    "airports.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:28:21.664523Z",
     "start_time": "2019-02-16T11:28:21.451120Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Table(name='temp', database=None, description=None, tableType='TEMPORARY', isTemporary=True)]\n",
      "[Table(name='airports', database=None, description=None, tableType='TEMPORARY', isTemporary=True), Table(name='temp', database=None, description=None, tableType='TEMPORARY', isTemporary=True)]\n"
     ]
    }
   ],
   "source": [
    "# Examine the tables in the catalog\n",
    "print(spark.catalog.listTables())\n",
    "\n",
    "# Add spark_temp to the catalog\n",
    "airports.createOrReplaceTempView('airports')\n",
    "\n",
    "# Examine the tables in the catalog again\n",
    "print(spark.catalog.listTables())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:28:24.951178Z",
     "start_time": "2019-02-16T11:28:24.832496Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+----------+------------+----+---+---+\n",
      "|faa|                name|       lat|         lon| alt| tz|dst|\n",
      "+---+--------------------+----------+------------+----+---+---+\n",
      "|04G|   Lansdowne Airport|41.1304722| -80.6195833|1044| -5|  A|\n",
      "|06A|Moton Field Munic...|32.4605722| -85.6800278| 264| -5|  A|\n",
      "|06C| Schaumburg Regional|41.9893408| -88.1012428| 801| -6|  A|\n",
      "|06N|     Randall Airport| 41.431912| -74.3915611| 523| -5|  A|\n",
      "|09J|Jekyll Island Air...|31.0744722| -81.4277778|  11| -4|  A|\n",
      "|0A9|Elizabethton Muni...|36.3712222| -82.1734167|1593| -4|  A|\n",
      "|0G6|Williams County A...|41.4673056| -84.5067778| 730| -5|  A|\n",
      "|0G7|Finger Lakes Regi...|42.8835647| -76.7812318| 492| -5|  A|\n",
      "|0P2|Shoestring Aviati...|39.7948244| -76.6471914|1000| -5|  U|\n",
      "|0S9|Jefferson County ...|48.0538086|-122.8106436| 108| -8|  A|\n",
      "+---+--------------------+----------+------------+----+---+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Don't change this query\n",
    "query = \"FROM airports SELECT * LIMIT 10\"\n",
    "\n",
    "# Get the first 10 rows of flights\n",
    "flights10 = spark.sql(query)\n",
    "\n",
    "# Show the results\n",
    "flights10.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Manipulating data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating columns\n",
    "\n",
    "In this chapter, you'll learn how to use the methods defined by Spark's DataFrame class to perform common data operations.\n",
    "\n",
    "Let's look at performing column-wise operations. In Spark you can do this using the .withColumn() method, which takes two arguments. First, a string with the name of your new column, and second the new column itself.\n",
    "\n",
    "The new column must be an object of class Column. Creating one of these is as easy as extracting a column from your DataFrame using df.colName.\n",
    "\n",
    "Updating a Spark DataFrame is somewhat different than working in pandas because the Spark DataFrame is immutable. This means that it can't be changed, and so columns can't be updated in place.\n",
    "\n",
    "Thus, all these methods return a new DataFrame. To overwrite the original DataFrame you must reassign the returned DataFrame using the method like so:\n",
    "\n",
    "df = df.withColumn(\"newCol\", df.oldCol + 1)\n",
    "\n",
    "The above code creates a DataFrame with the same columns as df plus a new column, newCol, where every entry is equal to the corresponding entry from oldCol, plus one.\n",
    "\n",
    "To overwrite an existing column, just pass the name of the column as the first argument!\n",
    "\n",
    "Remember, a SparkSession called spark is already in your workspace."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:34:09.603199Z",
     "start_time": "2019-02-16T11:34:09.384785Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Table(name='airports', database=None, description=None, tableType='TEMPORARY', isTemporary=True), Table(name='flights', database=None, description=None, tableType='TEMPORARY', isTemporary=True), Table(name='temp', database=None, description=None, tableType='TEMPORARY', isTemporary=True)]\n"
     ]
    }
   ],
   "source": [
    "# Don't change this file path\n",
    "file_path = \"flights_small.csv\"\n",
    "\n",
    "# Read in the airports data\n",
    "flights = spark.read.csv(file_path, header=True)\n",
    "\n",
    "# Add spark_temp to the catalog\n",
    "flights.createOrReplaceTempView('flights')\n",
    "\n",
    "# Examine the tables in the catalog again\n",
    "print(spark.catalog.listTables())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:34:13.318403Z",
     "start_time": "2019-02-16T11:34:13.181770Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+---+--------+---------+--------+---------+-------+-------+------+------+----+--------+--------+----+------+\n",
      "|year|month|day|dep_time|dep_delay|arr_time|arr_delay|carrier|tailnum|flight|origin|dest|air_time|distance|hour|minute|\n",
      "+----+-----+---+--------+---------+--------+---------+-------+-------+------+------+----+--------+--------+----+------+\n",
      "|2014|   12|  8|     658|       -7|     935|       -5|     VX| N846VA|  1780|   SEA| LAX|     132|     954|   6|    58|\n",
      "|2014|    1| 22|    1040|        5|    1505|        5|     AS| N559AS|   851|   SEA| HNL|     360|    2677|  10|    40|\n",
      "|2014|    3|  9|    1443|       -2|    1652|        2|     VX| N847VA|   755|   SEA| SFO|     111|     679|  14|    43|\n",
      "|2014|    4|  9|    1705|       45|    1839|       34|     WN| N360SW|   344|   PDX| SJC|      83|     569|  17|     5|\n",
      "|2014|    3|  9|     754|       -1|    1015|        1|     AS| N612AS|   522|   SEA| BUR|     127|     937|   7|    54|\n",
      "|2014|    1| 15|    1037|        7|    1352|        2|     WN| N646SW|    48|   PDX| DEN|     121|     991|  10|    37|\n",
      "|2014|    7|  2|     847|       42|    1041|       51|     WN| N422WN|  1520|   PDX| OAK|      90|     543|   8|    47|\n",
      "|2014|    5| 12|    1655|       -5|    1842|      -18|     VX| N361VA|   755|   SEA| SFO|      98|     679|  16|    55|\n",
      "|2014|    4| 19|    1236|       -4|    1508|       -7|     AS| N309AS|   490|   SEA| SAN|     135|    1050|  12|    36|\n",
      "|2014|   11| 19|    1812|       -3|    2352|       -4|     AS| N564AS|    26|   SEA| ORD|     198|    1721|  18|    12|\n",
      "|2014|   11|  8|    1653|       -2|    1924|       -1|     AS| N323AS|   448|   SEA| LAX|     130|     954|  16|    53|\n",
      "|2014|    8|  3|    1120|        0|    1415|        2|     AS| N305AS|   656|   SEA| PHX|     154|    1107|  11|    20|\n",
      "|2014|   10| 30|     811|       21|    1038|       29|     AS| N433AS|   608|   SEA| LAS|     127|     867|   8|    11|\n",
      "|2014|   11| 12|    2346|       -4|     217|      -28|     AS| N765AS|   121|   SEA| ANC|     183|    1448|  23|    46|\n",
      "|2014|   10| 31|    1314|       89|    1544|      111|     AS| N713AS|   306|   SEA| SFO|     129|     679|  13|    14|\n",
      "|2014|    1| 29|    2009|        3|    2159|        9|     UA| N27205|  1458|   PDX| SFO|      90|     550|  20|     9|\n",
      "|2014|   12| 17|    2015|       50|    2150|       41|     AS| N626AS|   368|   SEA| SMF|      76|     605|  20|    15|\n",
      "|2014|    8| 11|    1017|       -3|    1613|       -7|     WN| N8634A|   827|   SEA| MDW|     216|    1733|  10|    17|\n",
      "|2014|    1| 13|    2156|       -9|     607|      -15|     AS| N597AS|    24|   SEA| BOS|     290|    2496|  21|    56|\n",
      "|2014|    6|  5|    1733|      -12|    1945|      -10|     OO| N215AG|  3488|   PDX| BUR|     111|     817|  17|    33|\n",
      "+----+-----+---+--------+---------+--------+---------+-------+-------+------+------+----+--------+--------+----+------+\n",
      "only showing top 20 rows\n",
      "\n",
      "None\n"
     ]
    }
   ],
   "source": [
    "# Create the DataFrame flights\n",
    "flights = spark.table(\"flights\")\n",
    "\n",
    "# Show the head\n",
    "print(flights.show())\n",
    "\n",
    "# Add duration_hrs\n",
    "flights = flights.withColumn(\"duration_hrs\", flights.air_time/60)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Filtering Data\n",
    "\n",
    "Now that you have a bit of SQL know-how under your belt, it's easier to talk about the analogous operations using Spark DataFrames.\n",
    "\n",
    "Let's take a look at the .filter() method. As you might suspect, this is the Spark counterpart of SQL's WHERE clause. The .filter() method takes either a Spark Column of boolean (True/False) values or the WHERE clause of a SQL expression as a string.\n",
    "\n",
    "For example, the following two expressions will produce the same output:\n",
    "\n",
    "flights.filter(flights.air_time > 120).show()\n",
    "flights.filter(\"air_time > 120\").show()\n",
    "\n",
    "Remember, a SparkSession called spark is already in your workspace, along with the Spark DataFrame flights."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:35:03.159527Z",
     "start_time": "2019-02-16T11:35:02.835396Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+---+--------+---------+--------+---------+-------+-------+------+------+----+--------+--------+----+------+------------------+\n",
      "|year|month|day|dep_time|dep_delay|arr_time|arr_delay|carrier|tailnum|flight|origin|dest|air_time|distance|hour|minute|      duration_hrs|\n",
      "+----+-----+---+--------+---------+--------+---------+-------+-------+------+------+----+--------+--------+----+------+------------------+\n",
      "|2014|    1| 22|    1040|        5|    1505|        5|     AS| N559AS|   851|   SEA| HNL|     360|    2677|  10|    40|               6.0|\n",
      "|2014|    4| 19|    1236|       -4|    1508|       -7|     AS| N309AS|   490|   SEA| SAN|     135|    1050|  12|    36|              2.25|\n",
      "|2014|   11| 19|    1812|       -3|    2352|       -4|     AS| N564AS|    26|   SEA| ORD|     198|    1721|  18|    12|               3.3|\n",
      "|2014|    8|  3|    1120|        0|    1415|        2|     AS| N305AS|   656|   SEA| PHX|     154|    1107|  11|    20| 2.566666666666667|\n",
      "|2014|   11| 12|    2346|       -4|     217|      -28|     AS| N765AS|   121|   SEA| ANC|     183|    1448|  23|    46|              3.05|\n",
      "|2014|    8| 11|    1017|       -3|    1613|       -7|     WN| N8634A|   827|   SEA| MDW|     216|    1733|  10|    17|               3.6|\n",
      "|2014|    1| 13|    2156|       -9|     607|      -15|     AS| N597AS|    24|   SEA| BOS|     290|    2496|  21|    56| 4.833333333333333|\n",
      "|2014|    9| 26|     610|       -5|    1523|       65|     US| N127UW|   616|   SEA| PHL|     293|    2378|   6|    10| 4.883333333333334|\n",
      "|2014|   12|  4|     954|       -6|    1348|      -17|     HA| N395HA|    29|   SEA| OGG|     333|    2640|   9|    54|              5.55|\n",
      "|2014|    6|  4|    1115|        0|    1346|       -3|     AS| N461AS|   488|   SEA| SAN|     133|    1050|  11|    15| 2.216666666666667|\n",
      "|2014|    6| 26|    2054|       -1|    2318|       -6|     B6| N590JB|   907|   SEA| ANC|     179|    1448|  20|    54|2.9833333333333334|\n",
      "|2014|    6|  7|    1823|       -7|    2112|      -28|     AS| N512AS|   815|   SEA| LIH|     335|    2701|  18|    23| 5.583333333333333|\n",
      "|2014|    4| 30|     801|        1|    1757|       90|     AS| N407AS|    18|   SEA| MCO|     342|    2554|   8|     1|               5.7|\n",
      "|2014|   11| 29|     905|      155|    1655|      170|     DL| N824DN|  1598|   SEA| ATL|     229|    2182|   9|     5| 3.816666666666667|\n",
      "|2014|    6|  2|    2222|        7|      55|       15|     AS| N402AS|    99|   SEA| ANC|     190|    1448|  22|    22|3.1666666666666665|\n",
      "|2014|   11| 15|    1034|       -6|    1414|      -26|     AS| N589AS|   794|   SEA| ABQ|     139|    1180|  10|    34| 2.316666666666667|\n",
      "|2014|   10| 20|    1328|       -1|    1949|        4|     UA| N68805|  1212|   SEA| IAH|     228|    1874|  13|    28|               3.8|\n",
      "|2014|   12| 16|    1500|        0|    1906|       19|     US| N662AW|   500|   SEA| PHX|     151|    1107|  15|     0|2.5166666666666666|\n",
      "|2014|   11| 19|    1319|       -6|    1821|      -14|     DL| N309US|  2164|   PDX| MSP|     169|    1426|  13|    19| 2.816666666666667|\n",
      "|2014|    5| 21|     515|        0|     757|        0|     US| N172US|   593|   SEA| PHX|     143|    1107|   5|    15|2.3833333333333333|\n",
      "+----+-----+---+--------+---------+--------+---------+-------+-------+------+------+----+--------+--------+----+------+------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "None\n",
      "+----+-----+---+--------+---------+--------+---------+-------+-------+------+------+----+--------+--------+----+------+------------------+\n",
      "|year|month|day|dep_time|dep_delay|arr_time|arr_delay|carrier|tailnum|flight|origin|dest|air_time|distance|hour|minute|      duration_hrs|\n",
      "+----+-----+---+--------+---------+--------+---------+-------+-------+------+------+----+--------+--------+----+------+------------------+\n",
      "|2014|    1| 22|    1040|        5|    1505|        5|     AS| N559AS|   851|   SEA| HNL|     360|    2677|  10|    40|               6.0|\n",
      "|2014|    4| 19|    1236|       -4|    1508|       -7|     AS| N309AS|   490|   SEA| SAN|     135|    1050|  12|    36|              2.25|\n",
      "|2014|   11| 19|    1812|       -3|    2352|       -4|     AS| N564AS|    26|   SEA| ORD|     198|    1721|  18|    12|               3.3|\n",
      "|2014|    8|  3|    1120|        0|    1415|        2|     AS| N305AS|   656|   SEA| PHX|     154|    1107|  11|    20| 2.566666666666667|\n",
      "|2014|   11| 12|    2346|       -4|     217|      -28|     AS| N765AS|   121|   SEA| ANC|     183|    1448|  23|    46|              3.05|\n",
      "|2014|    8| 11|    1017|       -3|    1613|       -7|     WN| N8634A|   827|   SEA| MDW|     216|    1733|  10|    17|               3.6|\n",
      "|2014|    1| 13|    2156|       -9|     607|      -15|     AS| N597AS|    24|   SEA| BOS|     290|    2496|  21|    56| 4.833333333333333|\n",
      "|2014|    9| 26|     610|       -5|    1523|       65|     US| N127UW|   616|   SEA| PHL|     293|    2378|   6|    10| 4.883333333333334|\n",
      "|2014|   12|  4|     954|       -6|    1348|      -17|     HA| N395HA|    29|   SEA| OGG|     333|    2640|   9|    54|              5.55|\n",
      "|2014|    6|  4|    1115|        0|    1346|       -3|     AS| N461AS|   488|   SEA| SAN|     133|    1050|  11|    15| 2.216666666666667|\n",
      "|2014|    6| 26|    2054|       -1|    2318|       -6|     B6| N590JB|   907|   SEA| ANC|     179|    1448|  20|    54|2.9833333333333334|\n",
      "|2014|    6|  7|    1823|       -7|    2112|      -28|     AS| N512AS|   815|   SEA| LIH|     335|    2701|  18|    23| 5.583333333333333|\n",
      "|2014|    4| 30|     801|        1|    1757|       90|     AS| N407AS|    18|   SEA| MCO|     342|    2554|   8|     1|               5.7|\n",
      "|2014|   11| 29|     905|      155|    1655|      170|     DL| N824DN|  1598|   SEA| ATL|     229|    2182|   9|     5| 3.816666666666667|\n",
      "|2014|    6|  2|    2222|        7|      55|       15|     AS| N402AS|    99|   SEA| ANC|     190|    1448|  22|    22|3.1666666666666665|\n",
      "|2014|   11| 15|    1034|       -6|    1414|      -26|     AS| N589AS|   794|   SEA| ABQ|     139|    1180|  10|    34| 2.316666666666667|\n",
      "|2014|   10| 20|    1328|       -1|    1949|        4|     UA| N68805|  1212|   SEA| IAH|     228|    1874|  13|    28|               3.8|\n",
      "|2014|   12| 16|    1500|        0|    1906|       19|     US| N662AW|   500|   SEA| PHX|     151|    1107|  15|     0|2.5166666666666666|\n",
      "|2014|   11| 19|    1319|       -6|    1821|      -14|     DL| N309US|  2164|   PDX| MSP|     169|    1426|  13|    19| 2.816666666666667|\n",
      "|2014|    5| 21|     515|        0|     757|        0|     US| N172US|   593|   SEA| PHX|     143|    1107|   5|    15|2.3833333333333333|\n",
      "+----+-----+---+--------+---------+--------+---------+-------+-------+------+------+----+--------+--------+----+------+------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "None\n"
     ]
    }
   ],
   "source": [
    "# Filter flights with a SQL string\n",
    "long_flights1 = flights.filter(\"distance > 1000\")\n",
    "\n",
    "# Filter flights with a boolean column\n",
    "long_flights2 = flights.filter(flights.distance > 1000)\n",
    "\n",
    "# Examine the data to check they're equal\n",
    "print(long_flights1.show())\n",
    "print(long_flights2.show())\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:34:16.954855Z",
     "start_time": "2019-02-16T11:34:16.943857Z"
    }
   },
   "outputs": [],
   "source": [
    "# Import LogisticRegression\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "\n",
    "# Create a LogisticRegression Estimator\n",
    "lr = LogisticRegression()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:34:17.983053Z",
     "start_time": "2019-02-16T11:34:17.973084Z"
    }
   },
   "outputs": [],
   "source": [
    "# Import the evaluation submodule\n",
    "import pyspark.ml.evaluation as evals\n",
    "\n",
    "# Create a BinaryClassificationEvaluator\n",
    "evaluator = evals.BinaryClassificationEvaluator(metricName=\"areaUnderROC\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:28:28.145674Z",
     "start_time": "2019-02-16T11:28:28.140686Z"
    }
   },
   "outputs": [],
   "source": [
    "# Import the tuning submodule\n",
    "import pyspark.ml.tuning as tune\n",
    "\n",
    "# Create the parameter grid\n",
    "grid = tune.ParamGridBuilder()\n",
    "\n",
    "# Add the hyperparameter\n",
    "grid = grid.addGrid(lr.regParam, np.arange(0, .1, .01))\n",
    "grid = grid.addGrid(lr.elasticNetParam, [0, 1])\n",
    "\n",
    "# Build the grid\n",
    "grid = grid.build()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:28:29.012289Z",
     "start_time": "2019-02-16T11:28:29.007321Z"
    }
   },
   "outputs": [],
   "source": [
    "# Create the CrossValidator\n",
    "cv = tune.CrossValidator(estimator=lr,\n",
    "                         estimatorParamMaps=grid,\n",
    "                         evaluator=evaluator\n",
    "                         )\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Selecting\n",
    "\n",
    "The Spark variant of SQL's SELECT is the .select() method. This method takes multiple arguments - one for each column you want to select. These arguments can either be the column name as a string (one for each column) or a column object (using the df.colName syntax). When you pass a column object, you can perform operations like addition or subtraction on the column to change the data contained in it, much like inside .withColumn().\n",
    "\n",
    "The difference between .select() and .withColumn() methods is that .select() returns only the columns you specify, while .withColumn() returns all the columns of the DataFrame in addition to the one you defined. It's often a good idea to drop columns you don't need at the beginning of an operation so that you're not dragging around extra data as you're wrangling. In this case, you would use .select() and not .withColumn().\n",
    "\n",
    "Remember, a SparkSession called spark is already in your workspace, along with the Spark DataFrame flights."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:35:55.415393Z",
     "start_time": "2019-02-16T11:35:55.331618Z"
    }
   },
   "outputs": [],
   "source": [
    "# Select the first set of columns\n",
    "selected1 = flights.select(\"tailnum\", \"origin\", \"dest\")\n",
    "\n",
    "# Select the second set of columns\n",
    "temp = flights.select(flights.origin, flights.dest, flights.carrier)\n",
    "\n",
    "# Define first filter\n",
    "filterA = flights.origin == \"SEA\"\n",
    "\n",
    "# Define second filter\n",
    "filterB = flights.dest == \"PDX\"\n",
    "\n",
    "# Filter the data, first by filterA then by filterB\n",
    "selected2 = temp.filter(filterA).filter(filterB)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Selecting II\n",
    "\n",
    "Similar to SQL, you can also use the .select() method to perform column-wise operations. When you're selecting a column using the df.colName notation, you can perform any column operation and the .select() method will return the transformed column. For example,\n",
    "\n",
    "flights.select(flights.air_time/60)\n",
    "\n",
    "returns a column of flight durations in hours instead of minutes. You can also use the .alias() method to rename a column you're selecting. So if you wanted to .select() the column duration_hrs (which isn't in your DataFrame) you could do\n",
    "\n",
    "flights.select((flights.air_time/60).alias(\"duration_hrs\"))\n",
    "\n",
    "The equivalent Spark DataFrame method .selectExpr() takes SQL expressions as a string:\n",
    "\n",
    "flights.selectExpr(\"air_time/60 as duration_hrs\")\n",
    "\n",
    "with the SQL as keyword being equivalent to the .alias() method. To select multiple columns, you can pass multiple strings.\n",
    "\n",
    "Remember, a SparkSession called spark is already in your workspace, along with the Spark DataFrame flights."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:36:15.956312Z",
     "start_time": "2019-02-16T11:36:15.904452Z"
    }
   },
   "outputs": [],
   "source": [
    "# Define avg_speed\n",
    "avg_speed = (flights.distance/(flights.air_time/60)).alias(\"avg_speed\")\n",
    "\n",
    "# Select the correct columns\n",
    "speed1 = flights.select(\"origin\", \"dest\", \"tailnum\", avg_speed)\n",
    "\n",
    "# Create the same table using a SQL expression\n",
    "speed2 = flights.selectExpr(\"origin\", \"dest\", \"tailnum\", \"distance/(air_time/60) as avg_speed\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Aggregating\n",
    "\n",
    "All of the common aggregation methods, like .min(), .max(), and .count() are GroupedData methods. These are created by calling the .groupBy() DataFrame method. You'll learn exactly what that means in a few exercises. For now, all you have to do to use these functions is call that method on your DataFrame. For example, to find the minimum value of a column, col, in a DataFrame, df, you could do\n",
    "\n",
    "df.groupBy().min(\"col\").show()\n",
    "\n",
    "This creates a GroupedData object (so you can use the .min() method), then finds the minimum value in col, and returns it as a DataFrame.\n",
    "\n",
    "Now you're ready to do some aggregating of your own!\n",
    "\n",
    "A SparkSession called spark is already in your workspace, along with the Spark DataFrame flights."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:36:30.092987Z",
     "start_time": "2019-02-16T11:36:30.039109Z"
    }
   },
   "outputs": [
    {
     "ename": "AnalysisException",
     "evalue": "'\"distance\" is not a numeric column. Aggregation function can only be applied on a numeric column.;'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     62\u001b[0m         \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 63\u001b[1;33m             \u001b[1;32mreturn\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m*\u001b[0m\u001b[0ma\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     64\u001b[0m         \u001b[1;32mexcept\u001b[0m \u001b[0mpy4j\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mprotocol\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mPy4JJavaError\u001b[0m \u001b[1;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\py4j\\protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m    327\u001b[0m                     \u001b[1;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 328\u001b[1;33m                     format(target_id, \".\", name), value)\n\u001b[0m\u001b[0;32m    329\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling o358.min.\n: org.apache.spark.sql.AnalysisException: \"distance\" is not a numeric column. Aggregation function can only be applied on a numeric column.;\r\n\tat org.apache.spark.sql.RelationalGroupedDataset$$anonfun$4.apply(RelationalGroupedDataset.scala:101)\r\n\tat org.apache.spark.sql.RelationalGroupedDataset$$anonfun$4.apply(RelationalGroupedDataset.scala:98)\r\n\tat scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)\r\n\tat scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)\r\n\tat scala.collection.Iterator$class.foreach(Iterator.scala:891)\r\n\tat scala.collection.AbstractIterator.foreach(Iterator.scala:1334)\r\n\tat scala.collection.IterableLike$class.foreach(IterableLike.scala:72)\r\n\tat scala.collection.AbstractIterable.foreach(Iterable.scala:54)\r\n\tat scala.collection.TraversableLike$class.map(TraversableLike.scala:234)\r\n\tat scala.collection.AbstractTraversable.map(Traversable.scala:104)\r\n\tat org.apache.spark.sql.RelationalGroupedDataset.aggregateNumericColumns(RelationalGroupedDataset.scala:98)\r\n\tat org.apache.spark.sql.RelationalGroupedDataset.min(RelationalGroupedDataset.scala:284)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\r\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\r\n\tat java.lang.reflect.Method.invoke(Method.java:498)\r\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\r\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\r\n\tat py4j.Gateway.invoke(Gateway.java:282)\r\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\r\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\r\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\r\n\tat java.lang.Thread.run(Thread.java:748)\r\n",
      "\nDuring handling of the above exception, another exception occurred:\n",
      "\u001b[1;31mAnalysisException\u001b[0m                         Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-33-f39828568910>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m\u001b[0m\n\u001b[0;32m      1\u001b[0m \u001b[1;31m# Find the shortest flight from PDX in terms of distance\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 2\u001b[1;33m \u001b[0mflights\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mfilter\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mflights\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0morigin\u001b[0m \u001b[1;33m==\u001b[0m \u001b[1;34m\"PDX\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgroupBy\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mmin\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"distance\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mshow\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      3\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      4\u001b[0m \u001b[1;31m# Find the longest flight from SEA in terms of duration\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      5\u001b[0m \u001b[0mflights\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mfilter\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mflights\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0morigin\u001b[0m \u001b[1;33m==\u001b[0m \u001b[1;34m\"SEA\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgroupBy\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mmax\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"air_time\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mshow\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\group.py\u001b[0m in \u001b[0;36m_api\u001b[1;34m(self, *cols)\u001b[0m\n\u001b[0;32m     40\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0m_api\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m*\u001b[0m\u001b[0mcols\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     41\u001b[0m         \u001b[0mname\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m__name__\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 42\u001b[1;33m         \u001b[0mjdf\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mgetattr\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jgd\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mname\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0m_to_seq\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msql_ctx\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_sc\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mcols\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     43\u001b[0m         \u001b[1;32mreturn\u001b[0m \u001b[0mDataFrame\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mjdf\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msql_ctx\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     44\u001b[0m     \u001b[0m_api\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m__name__\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m__name__\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\py4j\\java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[1;34m(self, *args)\u001b[0m\n\u001b[0;32m   1255\u001b[0m         \u001b[0manswer\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1256\u001b[0m         return_value = get_return_value(\n\u001b[1;32m-> 1257\u001b[1;33m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[0;32m   1258\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1259\u001b[0m         \u001b[1;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     67\u001b[0m                                              e.java_exception.getStackTrace()))\n\u001b[0;32m     68\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'org.apache.spark.sql.AnalysisException: '\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 69\u001b[1;33m                 \u001b[1;32mraise\u001b[0m \u001b[0mAnalysisException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m': '\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     70\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'org.apache.spark.sql.catalyst.analysis'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     71\u001b[0m                 \u001b[1;32mraise\u001b[0m \u001b[0mAnalysisException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m': '\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mAnalysisException\u001b[0m: '\"distance\" is not a numeric column. Aggregation function can only be applied on a numeric column.;'"
     ]
    }
   ],
   "source": [
    "# Find the shortest flight from PDX in terms of distance\n",
    "flights.filter(flights.origin == \"PDX\").groupBy().min(\"distance\").show()\n",
    "\n",
    "# Find the longest flight from SEA in terms of duration\n",
    "flights.filter(flights.origin == \"SEA\").groupBy().max(\"air_time\").show()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Aggregating II\n",
    "\n",
    "To get you familiar with more of the built in aggregation methods, here's a few more exercises involving the flights table!\n",
    "\n",
    "Remember, a SparkSession called spark is already in your workspace, along with the Spark DataFrame flights."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:37:11.089942Z",
     "start_time": "2019-02-16T11:37:11.051048Z"
    }
   },
   "outputs": [
    {
     "ename": "AnalysisException",
     "evalue": "'\"air_time\" is not a numeric column. Aggregation function can only be applied on a numeric column.;'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     62\u001b[0m         \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 63\u001b[1;33m             \u001b[1;32mreturn\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m*\u001b[0m\u001b[0ma\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     64\u001b[0m         \u001b[1;32mexcept\u001b[0m \u001b[0mpy4j\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mprotocol\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mPy4JJavaError\u001b[0m \u001b[1;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\py4j\\protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m    327\u001b[0m                     \u001b[1;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 328\u001b[1;33m                     format(target_id, \".\", name), value)\n\u001b[0m\u001b[0;32m    329\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling o394.avg.\n: org.apache.spark.sql.AnalysisException: \"air_time\" is not a numeric column. Aggregation function can only be applied on a numeric column.;\r\n\tat org.apache.spark.sql.RelationalGroupedDataset$$anonfun$4.apply(RelationalGroupedDataset.scala:101)\r\n\tat org.apache.spark.sql.RelationalGroupedDataset$$anonfun$4.apply(RelationalGroupedDataset.scala:98)\r\n\tat scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)\r\n\tat scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)\r\n\tat scala.collection.Iterator$class.foreach(Iterator.scala:891)\r\n\tat scala.collection.AbstractIterator.foreach(Iterator.scala:1334)\r\n\tat scala.collection.IterableLike$class.foreach(IterableLike.scala:72)\r\n\tat scala.collection.AbstractIterable.foreach(Iterable.scala:54)\r\n\tat scala.collection.TraversableLike$class.map(TraversableLike.scala:234)\r\n\tat scala.collection.AbstractTraversable.map(Traversable.scala:104)\r\n\tat org.apache.spark.sql.RelationalGroupedDataset.aggregateNumericColumns(RelationalGroupedDataset.scala:98)\r\n\tat org.apache.spark.sql.RelationalGroupedDataset.avg(RelationalGroupedDataset.scala:272)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\r\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\r\n\tat java.lang.reflect.Method.invoke(Method.java:498)\r\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\r\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\r\n\tat py4j.Gateway.invoke(Gateway.java:282)\r\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\r\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\r\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\r\n\tat java.lang.Thread.run(Thread.java:748)\r\n",
      "\nDuring handling of the above exception, another exception occurred:\n",
      "\u001b[1;31mAnalysisException\u001b[0m                         Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-34-e6fc817cd92b>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m\u001b[0m\n\u001b[0;32m      1\u001b[0m \u001b[1;31m# Average duration of Delta flights\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 2\u001b[1;33m \u001b[0mflights\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mfilter\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mflights\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcarrier\u001b[0m \u001b[1;33m==\u001b[0m \u001b[1;34m\"DL\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mfilter\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mflights\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0morigin\u001b[0m \u001b[1;33m==\u001b[0m \u001b[1;34m\"SEA\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgroupBy\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mavg\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"air_time\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mshow\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      3\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      4\u001b[0m \u001b[1;31m# Total hours in the air\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      5\u001b[0m \u001b[0mflights\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mwithColumn\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"duration_hrs\"\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mflights\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mair_time\u001b[0m\u001b[1;33m/\u001b[0m\u001b[1;36m60\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgroupBy\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msum\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"duration_hrs\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mshow\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\group.py\u001b[0m in \u001b[0;36m_api\u001b[1;34m(self, *cols)\u001b[0m\n\u001b[0;32m     40\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0m_api\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m*\u001b[0m\u001b[0mcols\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     41\u001b[0m         \u001b[0mname\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m__name__\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 42\u001b[1;33m         \u001b[0mjdf\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mgetattr\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jgd\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mname\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0m_to_seq\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msql_ctx\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_sc\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mcols\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     43\u001b[0m         \u001b[1;32mreturn\u001b[0m \u001b[0mDataFrame\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mjdf\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msql_ctx\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     44\u001b[0m     \u001b[0m_api\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m__name__\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m__name__\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\py4j\\java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[1;34m(self, *args)\u001b[0m\n\u001b[0;32m   1255\u001b[0m         \u001b[0manswer\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1256\u001b[0m         return_value = get_return_value(\n\u001b[1;32m-> 1257\u001b[1;33m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[0;32m   1258\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1259\u001b[0m         \u001b[1;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     67\u001b[0m                                              e.java_exception.getStackTrace()))\n\u001b[0;32m     68\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'org.apache.spark.sql.AnalysisException: '\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 69\u001b[1;33m                 \u001b[1;32mraise\u001b[0m \u001b[0mAnalysisException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m': '\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     70\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'org.apache.spark.sql.catalyst.analysis'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     71\u001b[0m                 \u001b[1;32mraise\u001b[0m \u001b[0mAnalysisException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m': '\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mAnalysisException\u001b[0m: '\"air_time\" is not a numeric column. Aggregation function can only be applied on a numeric column.;'"
     ]
    }
   ],
   "source": [
    "# Average duration of Delta flights\n",
    "flights.filter(flights.carrier == \"DL\").filter(flights.origin == \"SEA\").groupBy().avg(\"air_time\").show()\n",
    "\n",
    "# Total hours in the air\n",
    "flights.withColumn(\"duration_hrs\", flights.air_time/60).groupBy().sum(\"duration_hrs\").show()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Grouping and Aggregating I\n",
    "\n",
    "Part of what makes aggregating so powerful is the addition of groups. PySpark has a whole class devoted to grouped data frames: pyspark.sql.GroupedData, which you saw in the last two exercises.\n",
    "\n",
    "You've learned how to create a grouped DataFrame by calling the .groupBy() method on a DataFrame with no arguments.\n",
    "\n",
    "Now you'll see that when you pass the name of one or more columns in your DataFrame to the .groupBy() method, the aggregation methods behave like when you use a GROUP BY statement in a SQL query!\n",
    "\n",
    "Remember, a SparkSession called spark is already in your workspace, along with the Spark DataFrame flights."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:37:25.024881Z",
     "start_time": "2019-02-16T11:37:23.336394Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+\n",
      "|tailnum|count|\n",
      "+-------+-----+\n",
      "| N442AS|   38|\n",
      "| N102UW|    2|\n",
      "| N36472|    4|\n",
      "| N38451|    4|\n",
      "| N73283|    4|\n",
      "| N513UA|    2|\n",
      "| N954WN|    5|\n",
      "| N388DA|    3|\n",
      "| N567AA|    1|\n",
      "| N516UA|    2|\n",
      "| N927DN|    1|\n",
      "| N8322X|    1|\n",
      "| N466SW|    1|\n",
      "|  N6700|    1|\n",
      "| N607AS|   45|\n",
      "| N622SW|    4|\n",
      "| N584AS|   31|\n",
      "| N914WN|    4|\n",
      "| N654AW|    2|\n",
      "| N336NW|    1|\n",
      "+-------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "ename": "AnalysisException",
     "evalue": "'\"air_time\" is not a numeric column. Aggregation function can only be applied on a numeric column.;'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     62\u001b[0m         \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 63\u001b[1;33m             \u001b[1;32mreturn\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m*\u001b[0m\u001b[0ma\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     64\u001b[0m         \u001b[1;32mexcept\u001b[0m \u001b[0mpy4j\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mprotocol\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mPy4JJavaError\u001b[0m \u001b[1;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\py4j\\protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m    327\u001b[0m                     \u001b[1;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 328\u001b[1;33m                     format(target_id, \".\", name), value)\n\u001b[0m\u001b[0;32m    329\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling o430.avg.\n: org.apache.spark.sql.AnalysisException: \"air_time\" is not a numeric column. Aggregation function can only be applied on a numeric column.;\r\n\tat org.apache.spark.sql.RelationalGroupedDataset$$anonfun$4.apply(RelationalGroupedDataset.scala:101)\r\n\tat org.apache.spark.sql.RelationalGroupedDataset$$anonfun$4.apply(RelationalGroupedDataset.scala:98)\r\n\tat scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)\r\n\tat scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)\r\n\tat scala.collection.Iterator$class.foreach(Iterator.scala:891)\r\n\tat scala.collection.AbstractIterator.foreach(Iterator.scala:1334)\r\n\tat scala.collection.IterableLike$class.foreach(IterableLike.scala:72)\r\n\tat scala.collection.AbstractIterable.foreach(Iterable.scala:54)\r\n\tat scala.collection.TraversableLike$class.map(TraversableLike.scala:234)\r\n\tat scala.collection.AbstractTraversable.map(Traversable.scala:104)\r\n\tat org.apache.spark.sql.RelationalGroupedDataset.aggregateNumericColumns(RelationalGroupedDataset.scala:98)\r\n\tat org.apache.spark.sql.RelationalGroupedDataset.avg(RelationalGroupedDataset.scala:272)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\r\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\r\n\tat java.lang.reflect.Method.invoke(Method.java:498)\r\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\r\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\r\n\tat py4j.Gateway.invoke(Gateway.java:282)\r\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\r\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\r\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\r\n\tat java.lang.Thread.run(Thread.java:748)\r\n",
      "\nDuring handling of the above exception, another exception occurred:\n",
      "\u001b[1;31mAnalysisException\u001b[0m                         Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-35-c446d57f1eff>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m\u001b[0m\n\u001b[0;32m      9\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     10\u001b[0m \u001b[1;31m# Average duration of flights from PDX and SEA\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 11\u001b[1;33m \u001b[0mby_origin\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mavg\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"air_time\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mshow\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\group.py\u001b[0m in \u001b[0;36m_api\u001b[1;34m(self, *cols)\u001b[0m\n\u001b[0;32m     40\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0m_api\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m*\u001b[0m\u001b[0mcols\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     41\u001b[0m         \u001b[0mname\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m__name__\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 42\u001b[1;33m         \u001b[0mjdf\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mgetattr\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jgd\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mname\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0m_to_seq\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msql_ctx\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_sc\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mcols\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     43\u001b[0m         \u001b[1;32mreturn\u001b[0m \u001b[0mDataFrame\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mjdf\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msql_ctx\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     44\u001b[0m     \u001b[0m_api\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m__name__\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m__name__\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\py4j\\java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[1;34m(self, *args)\u001b[0m\n\u001b[0;32m   1255\u001b[0m         \u001b[0manswer\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1256\u001b[0m         return_value = get_return_value(\n\u001b[1;32m-> 1257\u001b[1;33m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[0;32m   1258\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1259\u001b[0m         \u001b[1;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     67\u001b[0m                                              e.java_exception.getStackTrace()))\n\u001b[0;32m     68\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'org.apache.spark.sql.AnalysisException: '\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 69\u001b[1;33m                 \u001b[1;32mraise\u001b[0m \u001b[0mAnalysisException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m': '\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     70\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'org.apache.spark.sql.catalyst.analysis'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     71\u001b[0m                 \u001b[1;32mraise\u001b[0m \u001b[0mAnalysisException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m': '\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mAnalysisException\u001b[0m: '\"air_time\" is not a numeric column. Aggregation function can only be applied on a numeric column.;'"
     ]
    }
   ],
   "source": [
    "# Group by tailnum\n",
    "by_plane = flights.groupBy(\"tailnum\")\n",
    "\n",
    "# Number of flights each plane made\n",
    "by_plane.count().show()\n",
    "\n",
    "# Group by origin\n",
    "by_origin = flights.groupBy(\"origin\")\n",
    "\n",
    "# Average duration of flights from PDX and SEA\n",
    "by_origin.avg(\"air_time\").show()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Grouping and Aggregating II\n",
    "\n",
    "In addition to the GroupedData methods you've already seen, there is also the .agg() method. This method lets you pass an aggregate column expression that uses any of the aggregate functions from the pyspark.sql.functions submodule.\n",
    "\n",
    "This submodule contains many useful functions for computing things like standard deviations. All the aggregation functions in this submodule take the name of a column in a GroupedData table.\n",
    "\n",
    "Remember, a SparkSession called spark is already in your workspace, along with the Spark DataFrame flights. The grouped DataFrames you created in the last exercise are also in your workspace."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:37:36.217804Z",
     "start_time": "2019-02-16T11:37:36.177899Z"
    }
   },
   "outputs": [
    {
     "ename": "AnalysisException",
     "evalue": "'\"dep_delay\" is not a numeric column. Aggregation function can only be applied on a numeric column.;'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     62\u001b[0m         \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 63\u001b[1;33m             \u001b[1;32mreturn\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m*\u001b[0m\u001b[0ma\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     64\u001b[0m         \u001b[1;32mexcept\u001b[0m \u001b[0mpy4j\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mprotocol\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mPy4JJavaError\u001b[0m \u001b[1;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\py4j\\protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m    327\u001b[0m                     \u001b[1;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 328\u001b[1;33m                     format(target_id, \".\", name), value)\n\u001b[0m\u001b[0;32m    329\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling o462.avg.\n: org.apache.spark.sql.AnalysisException: \"dep_delay\" is not a numeric column. Aggregation function can only be applied on a numeric column.;\r\n\tat org.apache.spark.sql.RelationalGroupedDataset$$anonfun$4.apply(RelationalGroupedDataset.scala:101)\r\n\tat org.apache.spark.sql.RelationalGroupedDataset$$anonfun$4.apply(RelationalGroupedDataset.scala:98)\r\n\tat scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)\r\n\tat scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)\r\n\tat scala.collection.Iterator$class.foreach(Iterator.scala:891)\r\n\tat scala.collection.AbstractIterator.foreach(Iterator.scala:1334)\r\n\tat scala.collection.IterableLike$class.foreach(IterableLike.scala:72)\r\n\tat scala.collection.AbstractIterable.foreach(Iterable.scala:54)\r\n\tat scala.collection.TraversableLike$class.map(TraversableLike.scala:234)\r\n\tat scala.collection.AbstractTraversable.map(Traversable.scala:104)\r\n\tat org.apache.spark.sql.RelationalGroupedDataset.aggregateNumericColumns(RelationalGroupedDataset.scala:98)\r\n\tat org.apache.spark.sql.RelationalGroupedDataset.avg(RelationalGroupedDataset.scala:272)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\r\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\r\n\tat java.lang.reflect.Method.invoke(Method.java:498)\r\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\r\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\r\n\tat py4j.Gateway.invoke(Gateway.java:282)\r\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\r\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\r\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\r\n\tat java.lang.Thread.run(Thread.java:748)\r\n",
      "\nDuring handling of the above exception, another exception occurred:\n",
      "\u001b[1;31mAnalysisException\u001b[0m                         Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-36-78c5b9b511f0>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m\u001b[0m\n\u001b[0;32m      6\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      7\u001b[0m \u001b[1;31m# Average departure delay by month and destination\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 8\u001b[1;33m \u001b[0mby_month_dest\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mavg\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"dep_delay\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mshow\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      9\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     10\u001b[0m \u001b[1;31m# Standard deviation\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\group.py\u001b[0m in \u001b[0;36m_api\u001b[1;34m(self, *cols)\u001b[0m\n\u001b[0;32m     40\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0m_api\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m*\u001b[0m\u001b[0mcols\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     41\u001b[0m         \u001b[0mname\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m__name__\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 42\u001b[1;33m         \u001b[0mjdf\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mgetattr\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jgd\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mname\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0m_to_seq\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msql_ctx\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_sc\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mcols\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     43\u001b[0m         \u001b[1;32mreturn\u001b[0m \u001b[0mDataFrame\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mjdf\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msql_ctx\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     44\u001b[0m     \u001b[0m_api\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m__name__\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m__name__\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\py4j\\java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[1;34m(self, *args)\u001b[0m\n\u001b[0;32m   1255\u001b[0m         \u001b[0manswer\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1256\u001b[0m         return_value = get_return_value(\n\u001b[1;32m-> 1257\u001b[1;33m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[0;32m   1258\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1259\u001b[0m         \u001b[1;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m~\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     67\u001b[0m                                              e.java_exception.getStackTrace()))\n\u001b[0;32m     68\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'org.apache.spark.sql.AnalysisException: '\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 69\u001b[1;33m                 \u001b[1;32mraise\u001b[0m \u001b[0mAnalysisException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m': '\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     70\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'org.apache.spark.sql.catalyst.analysis'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     71\u001b[0m                 \u001b[1;32mraise\u001b[0m \u001b[0mAnalysisException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m': '\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mAnalysisException\u001b[0m: '\"dep_delay\" is not a numeric column. Aggregation function can only be applied on a numeric column.;'"
     ]
    }
   ],
   "source": [
    "# Import pyspark.sql.functions as F\n",
    "import pyspark.sql.functions as F\n",
    "\n",
    "# Group by month and dest\n",
    "by_month_dest = flights.groupBy(\"month\", \"dest\")\n",
    "\n",
    "# Average departure delay by month and destination\n",
    "by_month_dest.avg(\"dep_delay\").show()\n",
    "\n",
    "# Standard deviation\n",
    "by_month_dest.agg(F.stddev(\"dep_delay\")).show()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Joining\n",
    "\n",
    "Another very common data operation is the join. Joins are a whole topic unto themselves, so in this course we'll just look at simple joins. If you'd like to learn more about joins, you can take a look here.\n",
    "\n",
    "A join will combine two different tables along a column that they share. This column is called the key. Examples of keys here include the tailnum and carrier columns from the flights table.\n",
    "\n",
    "For example, suppose that you want to know more information about the plane that flew a flight than just the tail number. This information isn't in the flights table because the same plane flies many different flights over the course of two years, so including this information in every row would result in a lot of duplication. To avoid this, you'd have a second table that has only one row for each plane and whose columns list all the information about the plane, including its tail number. You could call this table planes\n",
    "\n",
    "When you join the flights table to this table of airplane information, you're adding all the columns from the planes table to the flights table. To fill these columns with information, you'll look at the tail number from the flights table and find the matching one in the planes table, and then use that row to fill out all the new columns.\n",
    "\n",
    "Now you'll have a much bigger table than before, but now every row has all information about the plane that flew that flight!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Joining II\n",
    "\n",
    "In PySpark, joins are performed using the DataFrame method .join(). This method takes three arguments. The first is the second DataFrame that you want to join with the first one. The second argument, on, is the name of the key column(s) as a string. The names of the key column(s) must be the same in each table. The third argument, how, specifies the kind of join to perform. In this course we'll always use the value how=\"leftouter\".\n",
    "\n",
    "The flights dataset and a new dataset called airports are already in your workspace."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:38:12.540127Z",
     "start_time": "2019-02-16T11:38:12.116259Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+----------------+-----------------+----+---+---+\n",
      "|faa|                name|             lat|              lon| alt| tz|dst|\n",
      "+---+--------------------+----------------+-----------------+----+---+---+\n",
      "|04G|   Lansdowne Airport|      41.1304722|      -80.6195833|1044| -5|  A|\n",
      "|06A|Moton Field Munic...|      32.4605722|      -85.6800278| 264| -5|  A|\n",
      "|06C| Schaumburg Regional|      41.9893408|      -88.1012428| 801| -6|  A|\n",
      "|06N|     Randall Airport|       41.431912|      -74.3915611| 523| -5|  A|\n",
      "|09J|Jekyll Island Air...|      31.0744722|      -81.4277778|  11| -4|  A|\n",
      "|0A9|Elizabethton Muni...|      36.3712222|      -82.1734167|1593| -4|  A|\n",
      "|0G6|Williams County A...|      41.4673056|      -84.5067778| 730| -5|  A|\n",
      "|0G7|Finger Lakes Regi...|      42.8835647|      -76.7812318| 492| -5|  A|\n",
      "|0P2|Shoestring Aviati...|      39.7948244|      -76.6471914|1000| -5|  U|\n",
      "|0S9|Jefferson County ...|      48.0538086|     -122.8106436| 108| -8|  A|\n",
      "|0W3|Harford County Ai...|      39.5668378|      -76.2024028| 409| -5|  A|\n",
      "|10C|  Galt Field Airport|      42.4028889|      -88.3751111| 875| -6|  U|\n",
      "|17G|Port Bucyrus-Craw...|      40.7815556|      -82.9748056|1003| -5|  A|\n",
      "|19A|Jackson County Ai...|      34.1758638|      -83.5615972| 951| -4|  U|\n",
      "|1A3|Martin Campbell F...|      35.0158056|      -84.3468333|1789| -4|  A|\n",
      "|1B9| Mansfield Municipal|      42.0001331|      -71.1967714| 122| -5|  A|\n",
      "|1C9|Frazier Lake Airpark|54.0133333333333|-124.768333333333| 152| -8|  A|\n",
      "|1CS|Clow Internationa...|      41.6959744|      -88.1292306| 670| -6|  U|\n",
      "|1G3|  Kent State Airport|      41.1513889|      -81.4151111|1134| -4|  A|\n",
      "|1OH|     Fortman Airport|      40.5553253|      -84.3866186| 885| -5|  U|\n",
      "+---+--------------------+----------------+-----------------+----+---+---+\n",
      "only showing top 20 rows\n",
      "\n",
      "None\n",
      "+----+----+-----+---+--------+---------+--------+---------+-------+-------+------+------+--------+--------+----+------+------------------+--------------------+---------+-----------+----+---+---+\n",
      "|dest|year|month|day|dep_time|dep_delay|arr_time|arr_delay|carrier|tailnum|flight|origin|air_time|distance|hour|minute|      duration_hrs|                name|      lat|        lon| alt| tz|dst|\n",
      "+----+----+-----+---+--------+---------+--------+---------+-------+-------+------+------+--------+--------+----+------+------------------+--------------------+---------+-----------+----+---+---+\n",
      "| LAX|2014|   12|  8|     658|       -7|     935|       -5|     VX| N846VA|  1780|   SEA|     132|     954|   6|    58|               2.2|    Los Angeles Intl|33.942536|-118.408075| 126| -8|  A|\n",
      "| HNL|2014|    1| 22|    1040|        5|    1505|        5|     AS| N559AS|   851|   SEA|     360|    2677|  10|    40|               6.0|       Honolulu Intl|21.318681|-157.922428|  13|-10|  N|\n",
      "| SFO|2014|    3|  9|    1443|       -2|    1652|        2|     VX| N847VA|   755|   SEA|     111|     679|  14|    43|              1.85|  San Francisco Intl|37.618972|-122.374889|  13| -8|  A|\n",
      "| SJC|2014|    4|  9|    1705|       45|    1839|       34|     WN| N360SW|   344|   PDX|      83|     569|  17|     5|1.3833333333333333|Norman Y Mineta S...|  37.3626|-121.929022|  62| -8|  A|\n",
      "| BUR|2014|    3|  9|     754|       -1|    1015|        1|     AS| N612AS|   522|   SEA|     127|     937|   7|    54|2.1166666666666667|            Bob Hope|34.200667|-118.358667| 778| -8|  A|\n",
      "| DEN|2014|    1| 15|    1037|        7|    1352|        2|     WN| N646SW|    48|   PDX|     121|     991|  10|    37|2.0166666666666666|         Denver Intl|39.861656|-104.673178|5431| -7|  A|\n",
      "| OAK|2014|    7|  2|     847|       42|    1041|       51|     WN| N422WN|  1520|   PDX|      90|     543|   8|    47|               1.5|Metropolitan Oakl...|37.721278|-122.220722|   9| -8|  A|\n",
      "| SFO|2014|    5| 12|    1655|       -5|    1842|      -18|     VX| N361VA|   755|   SEA|      98|     679|  16|    55|1.6333333333333333|  San Francisco Intl|37.618972|-122.374889|  13| -8|  A|\n",
      "| SAN|2014|    4| 19|    1236|       -4|    1508|       -7|     AS| N309AS|   490|   SEA|     135|    1050|  12|    36|              2.25|      San Diego Intl|32.733556|-117.189667|  17| -8|  A|\n",
      "| ORD|2014|   11| 19|    1812|       -3|    2352|       -4|     AS| N564AS|    26|   SEA|     198|    1721|  18|    12|               3.3|  Chicago Ohare Intl|41.978603| -87.904842| 668| -6|  A|\n",
      "| LAX|2014|   11|  8|    1653|       -2|    1924|       -1|     AS| N323AS|   448|   SEA|     130|     954|  16|    53|2.1666666666666665|    Los Angeles Intl|33.942536|-118.408075| 126| -8|  A|\n",
      "| PHX|2014|    8|  3|    1120|        0|    1415|        2|     AS| N305AS|   656|   SEA|     154|    1107|  11|    20| 2.566666666666667|Phoenix Sky Harbo...|33.434278|-112.011583|1135| -7|  N|\n",
      "| LAS|2014|   10| 30|     811|       21|    1038|       29|     AS| N433AS|   608|   SEA|     127|     867|   8|    11|2.1166666666666667|      Mc Carran Intl|36.080056| -115.15225|2141| -8|  A|\n",
      "| ANC|2014|   11| 12|    2346|       -4|     217|      -28|     AS| N765AS|   121|   SEA|     183|    1448|  23|    46|              3.05|Ted Stevens Ancho...|61.174361|-149.996361| 152| -9|  A|\n",
      "| SFO|2014|   10| 31|    1314|       89|    1544|      111|     AS| N713AS|   306|   SEA|     129|     679|  13|    14|              2.15|  San Francisco Intl|37.618972|-122.374889|  13| -8|  A|\n",
      "| SFO|2014|    1| 29|    2009|        3|    2159|        9|     UA| N27205|  1458|   PDX|      90|     550|  20|     9|               1.5|  San Francisco Intl|37.618972|-122.374889|  13| -8|  A|\n",
      "| SMF|2014|   12| 17|    2015|       50|    2150|       41|     AS| N626AS|   368|   SEA|      76|     605|  20|    15|1.2666666666666666|     Sacramento Intl|38.695417|-121.590778|  27| -8|  A|\n",
      "| MDW|2014|    8| 11|    1017|       -3|    1613|       -7|     WN| N8634A|   827|   SEA|     216|    1733|  10|    17|               3.6| Chicago Midway Intl|41.785972| -87.752417| 620| -6|  A|\n",
      "| BOS|2014|    1| 13|    2156|       -9|     607|      -15|     AS| N597AS|    24|   SEA|     290|    2496|  21|    56| 4.833333333333333|General Edward La...|42.364347| -71.005181|  19| -5|  A|\n",
      "| BUR|2014|    6|  5|    1733|      -12|    1945|      -10|     OO| N215AG|  3488|   PDX|     111|     817|  17|    33|              1.85|            Bob Hope|34.200667|-118.358667| 778| -8|  A|\n",
      "+----+----+-----+---+--------+---------+--------+---------+-------+-------+------+------+--------+--------+----+------+------------------+--------------------+---------+-----------+----+---+---+\n",
      "only showing top 20 rows\n",
      "\n",
      "None\n"
     ]
    }
   ],
   "source": [
    "# Examine the data\n",
    "print(airports.show())\n",
    "\n",
    "# Rename the faa column\n",
    "airports = airports.withColumnRenamed(\"faa\", \"dest\")\n",
    "\n",
    "# Join the DataFrames\n",
    "flights_with_airports = flights.join(airports, on=\"dest\", how=\"leftouter\")\n",
    "\n",
    "# Examine the data again\n",
    "print(flights_with_airports.show())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Getting started with machine learning pipelines\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Join the DataFrames\n",
    "\n",
    "In the next two chapters you'll be working to build a model that predicts whether or not a flight will be delayed based on the flights data we've been working with. This model will also include information about the plane that flew the route, so the first step is to join the two tables: flights and planes!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:39:48.979421Z",
     "start_time": "2019-02-16T11:39:48.659263Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Table(name='airports', database=None, description=None, tableType='TEMPORARY', isTemporary=True), Table(name='flights', database=None, description=None, tableType='TEMPORARY', isTemporary=True), Table(name='planes', database=None, description=None, tableType='TEMPORARY', isTemporary=True), Table(name='temp', database=None, description=None, tableType='TEMPORARY', isTemporary=True)]\n"
     ]
    }
   ],
   "source": [
    "# Don't change this file path\n",
    "file_path = \"planes.csv\"\n",
    "\n",
    "# Read in the airports data\n",
    "planes = spark.read.csv(file_path, header=True)\n",
    "\n",
    "# Add spark_temp to the catalog\n",
    "planes.createOrReplaceTempView('planes')\n",
    "\n",
    "# Examine the tables in the catalog again\n",
    "print(spark.catalog.listTables())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:39:52.800654Z",
     "start_time": "2019-02-16T11:39:52.786671Z"
    }
   },
   "outputs": [],
   "source": [
    "# Rename year column\n",
    "planes = planes.withColumnRenamed(\"year\", \"plane_year\")\n",
    "\n",
    "# Join the DataFrames\n",
    "model_data = flights.join(planes, on=\"tailnum\", how=\"leftouter\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data types\n",
    "\n",
    "Good work! Before you get started modeling, it's important to know that Spark only handles numeric data. That means all of the columns in your DataFrame must be either integers or decimals (called 'doubles' in Spark).\n",
    "\n",
    "When we imported our data, we let Spark guess what kind of information each column held. Unfortunately, Spark doesn't always guess right and you can see that some of the columns in our DataFrame are strings containing numbers as opposed to actual numeric values.\n",
    "\n",
    "To remedy this, you can use the .cast() method in combination with the .withColumn() method. It's important to note that .cast() works on columns, while .withColumn() works on DataFrames.\n",
    "\n",
    "The only argument you need to pass to .cast() is the kind of value you want to create, in string form. For example, to create integers, you'll pass the argument \"integer\" and for decimal numbers you'll use \"double\".\n",
    "\n",
    "You can put this call to .cast() inside a call to .withColumn() to overwrite the already existing column, just like you did in the previous chapter!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## String to integer\n",
    "\n",
    "Now you'll use the .cast() method you learned in the previous exercise to convert all the appropriate columns from your DataFrame model_data to integers!\n",
    "\n",
    "To convert the type of a column using the .cast() method, you can write code like this:\n",
    "\n",
    "dataframe = dataframe.withColumn(\"col\", dataframe.col.cast(\"new_type\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:40:27.813891Z",
     "start_time": "2019-02-16T11:40:27.769010Z"
    }
   },
   "outputs": [],
   "source": [
    "# Cast the columns to integers\n",
    "model_data = model_data.withColumn(\"arr_delay\", model_data.arr_delay.cast(\"integer\"))\n",
    "model_data = model_data.withColumn(\"air_time\", model_data.air_time.cast(\"integer\"))\n",
    "model_data = model_data.withColumn(\"month\", model_data.month.cast(\"integer\"))\n",
    "model_data = model_data.withColumn(\"plane_year\", model_data.plane_year.cast(\"integer\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create a new column\n",
    "\n",
    "In the last exercise, you converted the column plane_year to an integer. This column holds the year each plane was manufactured. However, your model will use the planes' age, which is slightly different from the year it was made!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:40:39.754784Z",
     "start_time": "2019-02-16T11:40:39.742818Z"
    }
   },
   "outputs": [],
   "source": [
    "# Create the column plane_age\n",
    "model_data = model_data.withColumn(\"plane_age\", model_data.year - model_data.plane_year)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Making a Boolean\n",
    "\n",
    "Consider that you're modeling a yes or no question: is the flight late? However, your data contains the arrival delay in minutes for each flight. Thus, you'll need to create a boolean column which indicates whether the flight was late or not!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:40:50.911588Z",
     "start_time": "2019-02-16T11:40:50.880652Z"
    }
   },
   "outputs": [],
   "source": [
    "# Create is_late\n",
    "model_data = model_data.withColumn(\"is_late\", model_data.arr_delay > 0)\n",
    "\n",
    "# Convert to an integer\n",
    "model_data = model_data.withColumn(\"label\", model_data.is_late.cast(\"integer\"))\n",
    "\n",
    "# Remove missing values\n",
    "model_data = model_data.filter(\"arr_delay is not NULL and dep_delay is not NULL and air_time is not NULL and plane_year is not NULL\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Strings and factors\n",
    "\n",
    "As you know, Spark requires numeric data for modeling. So far this hasn't been an issue; even boolean columns can easily be converted to integers without any trouble. But you'll also be using the airline and the plane's destination as features in your model. These are coded as strings and there isn't any obvious way to convert them to a numeric data type.\n",
    "\n",
    "Fortunately, PySpark has functions for handling this built into the pyspark.ml.features submodule. You can create what are called 'one-hot vectors' to represent the carrier and the destination of each flight. A one-hot vector is a way of representing a categorical feature where every observation has a vector in which all elements are zero except for at most one element, which has a value of one (1).\n",
    "\n",
    "Each element in the vector corresponds to a level of the feature, so it's possible to tell what the right level is by seeing which element of the vector is equal to one (1).\n",
    "\n",
    "The first step to encoding your categorical feature is to create a StringIndexer. Members of this class are Estimators that take a DataFrame with a column of strings and map each unique string to a number. Then, the Estimator returns a Transformer that takes a DataFrame, attaches the mapping to it as metadata, and returns a new DataFrame with a numeric column corresponding to the string column.\n",
    "\n",
    "The second step is to encode this numeric column as a one-hot vector using a OneHotEncoder. This works exactly the same way as the StringIndexer by creating an Estimator and then a Transformer. The end result is a column that encodes your categorical feature as a vector that's suitable for machine learning routines!\n",
    "\n",
    "This may seem complicated, but don't worry! All you have to remember is that you need to create a StringIndexer and a OneHotEncoder, and the Pipeline will take care of the rest."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Carrier\n",
    "\n",
    "In this exercise you'll create a StringIndexer and a OneHotEncoder to code the carrier column. To do this, you'll call the class constructors with the arguments inputCol and outputCol.\n",
    "\n",
    "The inputCol is the name of the column you want to index or encode, and the outputCol is the name of the new column that the Transformer should create."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:45:54.711306Z",
     "start_time": "2019-02-16T11:45:54.698314Z"
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler\n",
    "\n",
    "# Create a StringIndexer\n",
    "carr_indexer = StringIndexer(inputCol=\"carrier\", outputCol=\"carrier_index\")\n",
    "\n",
    "# Create a OneHotEncoder\n",
    "carr_encoder = OneHotEncoder(inputCol=\"carrier_index\", outputCol=\"carrier_fact\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Destination\n",
    "\n",
    "Now you'll encode the dest column just like you did in the previous exercise."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:45:19.024258Z",
     "start_time": "2019-02-16T11:45:19.008318Z"
    }
   },
   "outputs": [],
   "source": [
    "# Create a StringIndexer\n",
    "dest_indexer = StringIndexer(inputCol=\"dest\", outputCol=\"dest_index\")\n",
    "\n",
    "# Create a OneHotEncoder\n",
    "dest_encoder = OneHotEncoder(inputCol=\"dest_index\", outputCol=\"dest_fact\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Assemble a vector\n",
    "\n",
    "Good work so far!\n",
    "\n",
    "The last step in the Pipeline is to combine all of the columns containing our features into a single column. This has to be done before modeling can take place because every Spark modeling routine expects the data to be in this form. You can do this by storing each of the values from a column as an entry in a vector. Then, from the model's point of view, every observation is a vector that contains all of the information about it and a label that tells the modeler what value that observation corresponds to.\n",
    "\n",
    "Because of this, the pyspark.ml.feature submodule contains a class called VectorAssembler. This Transformer takes all of the columns you specify and combines them into a new vector column."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:45:57.599680Z",
     "start_time": "2019-02-16T11:45:57.579755Z"
    }
   },
   "outputs": [],
   "source": [
    "# Make a VectorAssembler\n",
    "vec_assembler = VectorAssembler(inputCols=[\"month\", \"air_time\", \"carrier_fact\", \"dest_fact\", \"plane_age\"], outputCol=\"features\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create the pipeline\n",
    "\n",
    "You're finally ready to create a Pipeline!\n",
    "\n",
    "Pipeline is a class in the pyspark.ml module that combines all the Estimators and Transformers that you've already created. This lets you reuse the same modeling process over and over again by wrapping it up in one simple object. Neat, right?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:46:11.680740Z",
     "start_time": "2019-02-16T11:46:11.675750Z"
    }
   },
   "outputs": [],
   "source": [
    "# Import Pipeline\n",
    "from pyspark.ml import Pipeline\n",
    "\n",
    "# Make the pipeline\n",
    "flights_pipe = Pipeline(stages=[dest_indexer, dest_encoder, carr_indexer, carr_encoder, vec_assembler])\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Test vs Train\n",
    "\n",
    "After you've cleaned your data and gotten it ready for modeling, one of the most important steps is to split the data into a test set and a train set. After that, don't touch your test data until you think you have a good model! As you're building models and forming hypotheses, you can test them on your training data to get an idea of their performance.\n",
    "\n",
    "Once you've got your favorite model, you can see how well it predicts the new data in your test set. This never-before-seen data will give you a much more realistic idea of your model's performance in the real world when you're trying to predict or classify new data.\n",
    "\n",
    "In Spark it's important to make sure you split the data after all the transformations. This is because operations like StringIndexer don't always produce the same index even when given the same list of strings."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transform the data\n",
    "\n",
    "Hooray, now you're finally ready to pass your data through the Pipeline you created!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:46:40.597757Z",
     "start_time": "2019-02-16T11:46:39.262349Z"
    }
   },
   "outputs": [],
   "source": [
    "# Fit and transform the data\n",
    "piped_data = flights_pipe.fit(model_data).transform(model_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Split the data\n",
    "\n",
    "Now that you've done all your manipulations, the last step before modeling is to split the data!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:46:52.624265Z",
     "start_time": "2019-02-16T11:46:52.591328Z"
    }
   },
   "outputs": [],
   "source": [
    "# Split the data into training and test sets\n",
    "training, test = piped_data.randomSplit([.6, .4])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Model tuning and selection"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## What is logistic regression?\n",
    "\n",
    "The model you'll be fitting in this chapter is called a logistic regression. This model is very similar to a linear regression, but instead of predicting a numeric variable, it predicts the probability (between 0 and 1) of an event.\n",
    "\n",
    "To use this as a classification algorithm, all you have to do is assign a cutoff point to these probabilities. If the predicted probability is above the cutoff point, you classify that observation as a 'yes' (in this case, the flight being late), if it's below, you classify it as a 'no'!\n",
    "\n",
    "You'll tune this model by testing different values for several hyperparameters. A hyperparameter is just a value in the model that's not estimated from the data, but rather is supplied by the user to maximize performance. For this course it's not necessary to understand the mathematics behind all of these values - what's important is that you'll try out a few different choices and pick the best one."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create the modeler\n",
    "\n",
    "The Estimator you'll be using is a LogisticRegression from the pyspark.ml.classification submodule."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:47:42.985036Z",
     "start_time": "2019-02-16T11:47:42.975074Z"
    }
   },
   "outputs": [],
   "source": [
    "# Import LogisticRegression\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "\n",
    "# Create a LogisticRegression Estimator\n",
    "lr = LogisticRegression()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Cross validation\n",
    "\n",
    "In the next few exercises you'll be tuning your logistic regression model using a procedure called k-fold cross validation. This is a method of estimating the model's performance on unseen data (like your test DataFrame).\n",
    "\n",
    "It works by splitting the training data into a few different partitions. The exact number is up to you, but in this course you'll be using PySpark's default value of three. Once the data is split up, one of the partitions is set aside, and the model is fit to the others. Then the error is measured against the held out partition. This is repeated for each of the partitions, so that every block of data is held out and used as a test set exactly once. Then the error on each of the partitions is averaged. This is called the cross validation error of the model, and is a good estimate of the actual error on the held out data.\n",
    "\n",
    "You'll be using cross validation to choose the hyperparameters by creating a grid of the possible pairs of values for the two hyperparameters, elasticNetParam and regParam, and using the cross validation error to compare all the different models so you can choose the best one!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create the evaluator\n",
    "\n",
    "The first thing you need when doing cross validation for model selection is a way to compare different models. Luckily, the pyspark.ml.evaluation submodule has classes for evaluating different kinds of models. Your model is a binary classification model, so you'll be using the BinaryClassificationEvaluator from the pyspark.ml.evaluation module.\n",
    "\n",
    "This evaluator calculates the area under the ROC. This is a metric that combines the two kinds of errors a binary classifier can make (false positives and false negatives) into a simple number. You'll learn more about this towards the end of the chapter!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:48:06.938836Z",
     "start_time": "2019-02-16T11:48:06.929853Z"
    }
   },
   "outputs": [],
   "source": [
    "# Import the evaluation submodule\n",
    "import pyspark.ml.evaluation as evals\n",
    "\n",
    "# Create a BinaryClassificationEvaluator\n",
    "evaluator = evals.BinaryClassificationEvaluator(metricName=\"areaUnderROC\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Make a grid\n",
    "\n",
    "Next, you need to create a grid of values to search over when looking for the optimal hyperparameters. The submodule pyspark.ml.tuning includes a class called ParamGridBuilder that does just that (maybe you're starting to notice a pattern here; PySpark has a submodule for just about everything!).\n",
    "\n",
    "You'll need to use the .addGrid() and .build() methods to create a grid that you can use for cross validation. The .addGrid() method takes a model parameter (an attribute of the model Estimator, lr, that you created a few exercises ago) and a list of values that you want to try. The .build() method takes no arguments, it just returns the grid that you'll use later."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:48:17.184878Z",
     "start_time": "2019-02-16T11:48:17.177892Z"
    }
   },
   "outputs": [],
   "source": [
    "# Import the tuning submodule\n",
    "import pyspark.ml.tuning as tune\n",
    "\n",
    "# Create the parameter grid\n",
    "grid = tune.ParamGridBuilder()\n",
    "\n",
    "# Add the hyperparameter\n",
    "grid = grid.addGrid(lr.regParam, np.arange(0, .1, .01))\n",
    "grid = grid.addGrid(lr.elasticNetParam, [0, 1])\n",
    "\n",
    "# Build the grid\n",
    "grid = grid.build()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Make the validator\n",
    "\n",
    "The submodule pyspark.ml.tuning also has a class called CrossValidator for performing cross validation. This Estimator takes the modeler you want to fit, the grid of hyperparameters you created, and the evaluator you want to use to compare your models.\n",
    "\n",
    "The submodule pyspark.ml.tune has already been imported as tune. You'll create the CrossValidator by passing it the logistic regression Estimator lr, the parameter grid, and the evaluator you created in the previous exercises."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:48:35.284411Z",
     "start_time": "2019-02-16T11:48:35.280438Z"
    }
   },
   "outputs": [],
   "source": [
    "# Create the CrossValidator\n",
    "cv = tune.CrossValidator(estimator=lr,\n",
    "                         estimatorParamMaps=grid,\n",
    "                         evaluator=evaluator\n",
    "                         )\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Fit the model(s)\n",
    "\n",
    "You're finally ready to fit the models and select the best one!\n",
    "\n",
    "Unfortunately, cross validation is a very computationally intensive procedure. Fitting all the models would take too long on DataCamp.\n",
    "\n",
    "To do this locally you would use the code\n",
    "\n",
    "    # Fit cross validation models\n",
    "    models = cv.fit(training)\n",
    "\n",
    "    # Extract the best model\n",
    "    best_lr = models.bestModel\n",
    "\n",
    "Remember, the training data is called training and you're using lr to fit a logistic regression model. Cross validation selected the parameter values regParam=0 and elasticNetParam=0 as being the best. These are the default values, so you don't need to do anything else with lr before fitting the model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:49:23.090501Z",
     "start_time": "2019-02-16T11:49:18.828812Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "LogisticRegressionModel: uid = LogisticRegression_5489d7442292, numClasses = 2, numFeatures = 81\n"
     ]
    }
   ],
   "source": [
    "# Call lr.fit()\n",
    "best_lr = lr.fit(training)\n",
    "\n",
    "# Print best_lr\n",
    "print(best_lr)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Evaluating binary classifiers\n",
    "\n",
    "For this course we'll be using a common metric for binary classification algorithms call the AUC, or area under the curve. In this case, the curve is the ROC, or receiver operating curve. The details of what these things actually measure isn't important for this course. All you need to know is that for our purposes, the closer the AUC is to one (1), the better the model is!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Evaluate the model\n",
    "\n",
    "Remember the test data that you set aside waaaaaay back in chapter 3? It's finally time to test your model on it! You can use the same evaluator you made to fit the model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-16T11:49:44.207903Z",
     "start_time": "2019-02-16T11:49:42.988166Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.6859503860421474\n"
     ]
    }
   ],
   "source": [
    "# Use the model to predict the test set\n",
    "test_results = best_lr.transform(test)\n",
    "\n",
    "# Evaluate the predictions\n",
    "print(evaluator.evaluate(test_results))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Conclusion\n",
    "\n",
    "Congrats on making it to the end of the course! You went from knowing nothing about Spark to doing advanced machine learning. Cool huh?\n",
    "\n",
    "The next steps are learning how to create large scale Spark clusters and manage and submit jobs so that you can use models in the real world.\n",
    "\n",
    "And remember, Spark is still being actively developed, so there's new features coming all the time!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.7"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": true
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}