W3-06-PySpark-Tutorial-Dataframes.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark Tutorial - Dataframes\n",
    "<div>\n",
    " <h2> CSCI 4283 / 5253 \n",
    "  <IMG SRC=\"https://www.colorado.edu/cs/profiles/express/themes/cuspirit/logo.png\" WIDTH=50 ALIGN=\"right\"/> </h2>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Up to now, we've see the RDD interface to PySpark. The RDD is a building block for more capable data structures such as the **dataframe** and **database**. These data structures are part of the [PySpark SQL library](https://spark.apache.org/docs/latest/api/python/pyspark.sql.html) which, as the name implies, is influenced by standard SQL practices and queries.\n",
    "\n",
    "The PySpark library has the **dataframe API**, but it does not support the **database API** -- that's only accessible via the Scala and Java libraries and through SQL queries.\n",
    "\n",
    "The **database** is effectively an SQL relation -- i.e. rows and columns with a specific schema. The **dataframe** takes a little futher and constructs a labeled dataframe similar to the [Python Pandas](https://pandas.pydata.org/) interface or the [R dplyr](https://cran.r-project.org/web/packages/dplyr/vignettes/dplyr.html) interface for R."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql import SparkSession\n",
    "import pyspark.sql.functions\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import operator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We're going to use an airline information database as the example. You can download extended versions of the database [at this Dept. of Transportation website](https://www.transtats.bts.gov/DL_SelectFields.asp), but the data we're using is distributed with the course notes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As with the RDD interface, we need a \"context\" to a remote machine. The [Spark SQL tutorial](https://spark.apache.org/docs/latest/sql-getting-started.html) has some information on this, but for complete information you need to look at the [Spark API documentation.](https://spark.apache.org/docs/latest/api/python/)\n",
    "\n",
    "In this example, we're creating a local session (i.e. CPU's on JupyterHub)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Python Spark SQL basic example\") \\\n",
    "    .master(\"local[*]\")\\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There are many ways to load data, including HDFS, a format called [Parquet](http://parquet.apache.org/), CSV files and so on. We'll use a compressed CSV file of the airline data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "flights = spark.read.load('airline-ontime-reporting.csv.gz',\n",
    "            format=\"csv\", sep=\",\", header=True,\n",
    "            compression=\"gzip\",\n",
    "            inferSchema=\"true\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The dataframe has a **schema** or type for each entry. All entries must have the same type or we'll see operations fail. In this example, we have asked that the schema be inferred -- this usually works, but if it doesn't we may need to take some extra steps (see below)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- YEAR: integer (nullable = true)\n",
      " |-- MONTH: integer (nullable = true)\n",
      " |-- DAY_OF_MONTH: integer (nullable = true)\n",
      " |-- DAY_OF_WEEK: integer (nullable = true)\n",
      " |-- OP_UNIQUE_CARRIER: string (nullable = true)\n",
      " |-- ORIGIN_AIRPORT_ID: integer (nullable = true)\n",
      " |-- ORIGIN_AIRPORT_SEQ_ID: integer (nullable = true)\n",
      " |-- ORIGIN_CITY_MARKET_ID: integer (nullable = true)\n",
      " |-- ORIGIN: string (nullable = true)\n",
      " |-- ORIGIN_CITY_NAME: string (nullable = true)\n",
      " |-- DEST_AIRPORT_ID: integer (nullable = true)\n",
      " |-- DEST_AIRPORT_SEQ_ID: integer (nullable = true)\n",
      " |-- DEST_CITY_MARKET_ID: integer (nullable = true)\n",
      " |-- DEST: string (nullable = true)\n",
      " |-- DEST_CITY_NAME: string (nullable = true)\n",
      " |-- CRS_DEP_TIME: integer (nullable = true)\n",
      " |-- DEP_DELAY: double (nullable = true)\n",
      " |-- ARR_TIME: integer (nullable = true)\n",
      " |-- ARR_DELAY: double (nullable = true)\n",
      " |-- CANCELLED: double (nullable = true)\n",
      " |-- CARRIER_DELAY: double (nullable = true)\n",
      " |-- WEATHER_DELAY: double (nullable = true)\n",
      " |-- LATE_AIRCRAFT_DELAY: double (nullable = true)\n",
      " |-- _c23: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "flights.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['YEAR',\n",
       " 'MONTH',\n",
       " 'DAY_OF_MONTH',\n",
       " 'DAY_OF_WEEK',\n",
       " 'OP_UNIQUE_CARRIER',\n",
       " 'ORIGIN_AIRPORT_ID',\n",
       " 'ORIGIN_AIRPORT_SEQ_ID',\n",
       " 'ORIGIN_CITY_MARKET_ID',\n",
       " 'ORIGIN',\n",
       " 'ORIGIN_CITY_NAME',\n",
       " 'DEST_AIRPORT_ID',\n",
       " 'DEST_AIRPORT_SEQ_ID',\n",
       " 'DEST_CITY_MARKET_ID',\n",
       " 'DEST',\n",
       " 'DEST_CITY_NAME',\n",
       " 'CRS_DEP_TIME',\n",
       " 'DEP_DELAY',\n",
       " 'ARR_TIME',\n",
       " 'ARR_DELAY',\n",
       " 'CANCELLED',\n",
       " 'CARRIER_DELAY',\n",
       " 'WEATHER_DELAY',\n",
       " 'LATE_AIRCRAFT_DELAY',\n",
       " '_c23']"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "flights.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "There are 24 columns and  583985 rows\n"
     ]
    }
   ],
   "source": [
    "print(\"There are\", len(flights.columns), \"columns and \", flights.count(), \"rows\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('YEAR', 'int'),\n",
       " ('MONTH', 'int'),\n",
       " ('DAY_OF_MONTH', 'int'),\n",
       " ('DAY_OF_WEEK', 'int'),\n",
       " ('OP_UNIQUE_CARRIER', 'string'),\n",
       " ('ORIGIN_AIRPORT_ID', 'int'),\n",
       " ('ORIGIN_AIRPORT_SEQ_ID', 'int'),\n",
       " ('ORIGIN_CITY_MARKET_ID', 'int'),\n",
       " ('ORIGIN', 'string'),\n",
       " ('ORIGIN_CITY_NAME', 'string'),\n",
       " ('DEST_AIRPORT_ID', 'int'),\n",
       " ('DEST_AIRPORT_SEQ_ID', 'int'),\n",
       " ('DEST_CITY_MARKET_ID', 'int'),\n",
       " ('DEST', 'string'),\n",
       " ('DEST_CITY_NAME', 'string'),\n",
       " ('CRS_DEP_TIME', 'int'),\n",
       " ('DEP_DELAY', 'double'),\n",
       " ('ARR_TIME', 'int'),\n",
       " ('ARR_DELAY', 'double'),\n",
       " ('CANCELLED', 'double'),\n",
       " ('CARRIER_DELAY', 'double'),\n",
       " ('WEATHER_DELAY', 'double'),\n",
       " ('LATE_AIRCRAFT_DELAY', 'double'),\n",
       " ('_c23', 'string')]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "flights.dtypes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The schema is inferred, but it can also be defined explicitly.\n",
    "\n",
    "Note that one column is labeled `_c23`, which is showing up as \"null\". Perhaps this is bad data import?\n",
    "\n",
    "Lets look at some of the values."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+--------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+----+\n",
      "|YEAR|MONTH|DAY_OF_MONTH|DAY_OF_WEEK|OP_UNIQUE_CARRIER|ORIGIN_AIRPORT_ID|ORIGIN_AIRPORT_SEQ_ID|ORIGIN_CITY_MARKET_ID|ORIGIN|ORIGIN_CITY_NAME|DEST_AIRPORT_ID|DEST_AIRPORT_SEQ_ID|DEST_CITY_MARKET_ID|DEST|DEST_CITY_NAME|CRS_DEP_TIME|DEP_DELAY|ARR_TIME|ARR_DELAY|CANCELLED|CARRIER_DELAY|WEATHER_DELAY|LATE_AIRCRAFT_DELAY|_c23|\n",
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+--------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+----+\n",
      "|2019|1    |19          |6          |9E               |13487            |1348702              |31650                |MSP   |Minneapolis, MN |11193          |1119302            |33105              |CVG |Cincinnati, OH|1556        |-10.0    |1832    |-25.0    |0.0      |null         |null         |null               |null|\n",
      "|2019|1    |20          |7          |9E               |13487            |1348702              |31650                |MSP   |Minneapolis, MN |11193          |1119302            |33105              |CVG |Cincinnati, OH|1556        |-4.0     |1825    |-37.0    |0.0      |null         |null         |null               |null|\n",
      "|2019|1    |21          |1          |9E               |13487            |1348702              |31650                |MSP   |Minneapolis, MN |11193          |1119302            |33105              |CVG |Cincinnati, OH|1556        |-9.0     |1845    |-17.0    |0.0      |null         |null         |null               |null|\n",
      "|2019|1    |22          |2          |9E               |13487            |1348702              |31650                |MSP   |Minneapolis, MN |11193          |1119302            |33105              |CVG |Cincinnati, OH|1556        |-4.0     |1839    |-23.0    |0.0      |null         |null         |null               |null|\n",
      "|2019|1    |23          |3          |9E               |13487            |1348702              |31650                |MSP   |Minneapolis, MN |11193          |1119302            |33105              |CVG |Cincinnati, OH|1556        |-6.0     |1850    |-12.0    |0.0      |null         |null         |null               |null|\n",
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+--------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "flights.show(5, truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's pull out the values in one column -- the `select` method can be used to produce a new dataframe with just that column as an entry."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+\n",
      "|_c23|\n",
      "+----+\n",
      "|null|\n",
      "|null|\n",
      "|null|\n",
      "|null|\n",
      "|null|\n",
      "+----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "flights.select('_c23').show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And we can slice out multiple columns, similar to Pandas. Again, this produces a new dataframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+\n",
      "|year|_c23|\n",
      "+----+----+\n",
      "|2019|null|\n",
      "|2019|null|\n",
      "|2019|null|\n",
      "|2019|null|\n",
      "|2019|null|\n",
      "+----+----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "flights.select(['year', '_c23']).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Alternatively, we can produce a\n",
    "[Column object which has its own methods](https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=column#pyspark.sql.Column). These are typically used in **column expressions** that produce indicies that can be used when selecting or filtering data.\n",
    "\n",
    "For example, let's find all the rows where the mystery `_c23` column is not null."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+--------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+----+\n",
      "|YEAR|MONTH|DAY_OF_MONTH|DAY_OF_WEEK|OP_UNIQUE_CARRIER|ORIGIN_AIRPORT_ID|ORIGIN_AIRPORT_SEQ_ID|ORIGIN_CITY_MARKET_ID|ORIGIN|ORIGIN_CITY_NAME|DEST_AIRPORT_ID|DEST_AIRPORT_SEQ_ID|DEST_CITY_MARKET_ID|DEST|DEST_CITY_NAME|CRS_DEP_TIME|DEP_DELAY|ARR_TIME|ARR_DELAY|CANCELLED|CARRIER_DELAY|WEATHER_DELAY|LATE_AIRCRAFT_DELAY|_c23|\n",
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+--------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+----+\n",
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+--------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "flights.filter( flights._c23.isNotNull()).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Hmm.... This liooks like all the values are null. We could confirm this by selecting the column and looking at the distinct elements."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+\n",
      "|_c23|\n",
      "+----+\n",
      "|null|\n",
      "+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "flights.select('_c23').distinct().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This this column is null, lets just drop it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "newFlights = flights.drop('_c23')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+--------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+\n",
      "|YEAR|MONTH|DAY_OF_MONTH|DAY_OF_WEEK|OP_UNIQUE_CARRIER|ORIGIN_AIRPORT_ID|ORIGIN_AIRPORT_SEQ_ID|ORIGIN_CITY_MARKET_ID|ORIGIN|ORIGIN_CITY_NAME|DEST_AIRPORT_ID|DEST_AIRPORT_SEQ_ID|DEST_CITY_MARKET_ID|DEST|DEST_CITY_NAME|CRS_DEP_TIME|DEP_DELAY|ARR_TIME|ARR_DELAY|CANCELLED|CARRIER_DELAY|WEATHER_DELAY|LATE_AIRCRAFT_DELAY|\n",
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+--------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+\n",
      "|2019|    1|          19|          6|               9E|            13487|              1348702|                31650|   MSP| Minneapolis, MN|          11193|            1119302|              33105| CVG|Cincinnati, OH|        1556|    -10.0|    1832|    -25.0|      0.0|         null|         null|               null|\n",
      "|2019|    1|          20|          7|               9E|            13487|              1348702|                31650|   MSP| Minneapolis, MN|          11193|            1119302|              33105| CVG|Cincinnati, OH|        1556|     -4.0|    1825|    -37.0|      0.0|         null|         null|               null|\n",
      "|2019|    1|          21|          1|               9E|            13487|              1348702|                31650|   MSP| Minneapolis, MN|          11193|            1119302|              33105| CVG|Cincinnati, OH|        1556|     -9.0|    1845|    -17.0|      0.0|         null|         null|               null|\n",
      "|2019|    1|          22|          2|               9E|            13487|              1348702|                31650|   MSP| Minneapolis, MN|          11193|            1119302|              33105| CVG|Cincinnati, OH|        1556|     -4.0|    1839|    -23.0|      0.0|         null|         null|               null|\n",
      "|2019|    1|          23|          3|               9E|            13487|              1348702|                31650|   MSP| Minneapolis, MN|          11193|            1119302|              33105| CVG|Cincinnati, OH|        1556|     -6.0|    1850|    -12.0|      0.0|         null|         null|               null|\n",
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+--------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "newFlights.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We often work with multiple columns of data in a dataframe. Some methods just use column names (corr, cov, crosstab, describe) and others can use column references, such as `newAir.ORIGIN`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There are also a number of methods that work on columns or column expressions -- we've been using `select` already.\n",
    "\n",
    "* `cube(*cols)`: column names (string) or column expressions or **both**.\n",
    "* `drop(*cols)`: ***a list of column names OR a single column expression.***\n",
    "* `groupBy(*cols)`: column name (string) or column expression or **both**.\n",
    "* `rollup(*cols)`: column name (string) or column expression or **both**.\n",
    "* `select(*cols)`: column name (string) or column expression or **both**.\n",
    "* `sort(*cols, **kwargs)`: column name (string) or column expression or **both**.\n",
    "* `sortWithinPartitions(*cols, **kwargs)`: column name (string) or column expression or **both**.\n",
    "* `orderBy(*cols, **kwargs)`: column name (string) or column expression or **both**.\n",
    "* `sampleBy(col, fractions, sed=None)`: a column name.\n",
    "* `toDF(*cols)`: **a list of column names (string).**\n",
    "* `withColumn(colName, col)`: `colName` refers to column name; `col` refers to a column expression.\n",
    "* `withColumnRenamed(existing, new)`: takes column names as arguments.\n",
    "* `filter(condition)`: ***condition** refers to a column expression that returns `types.BooleanType` of values. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For example, we could determine the number of flights from a specific origin by group by the flight origin and then counting the entries in each group:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+\n",
      "|ORIGIN|count|\n",
      "+------+-----+\n",
      "|   BGM|   61|\n",
      "|   PSE|   65|\n",
      "|   INL|   53|\n",
      "|   MSY| 4597|\n",
      "|   PPG|   11|\n",
      "+------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "newFlights.groupBy(newFlights.ORIGIN).count().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If we want to find the files from a _specific_ origin airport, we can `filter` those entries:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+---------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+\n",
      "|YEAR|MONTH|DAY_OF_MONTH|DAY_OF_WEEK|OP_UNIQUE_CARRIER|ORIGIN_AIRPORT_ID|ORIGIN_AIRPORT_SEQ_ID|ORIGIN_CITY_MARKET_ID|ORIGIN|ORIGIN_CITY_NAME|DEST_AIRPORT_ID|DEST_AIRPORT_SEQ_ID|DEST_CITY_MARKET_ID|DEST| DEST_CITY_NAME|CRS_DEP_TIME|DEP_DELAY|ARR_TIME|ARR_DELAY|CANCELLED|CARRIER_DELAY|WEATHER_DELAY|LATE_AIRCRAFT_DELAY|\n",
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+---------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+\n",
      "|2019|    1|           1|          2|               AA|            11292|              1129202|                30325|   DEN|      Denver, CO|          12892|            1289208|              32575| LAX|Los Angeles, CA|        2025|     -4.0|    2124|    -41.0|      0.0|         null|         null|               null|\n",
      "|2019|    1|           2|          3|               AA|            11292|              1129202|                30325|   DEN|      Denver, CO|          12892|            1289208|              32575| LAX|Los Angeles, CA|        2025|     -5.0|    2206|      1.0|      0.0|         null|         null|               null|\n",
      "|2019|    1|           3|          4|               AA|            11292|              1129202|                30325|   DEN|      Denver, CO|          12892|            1289208|              32575| LAX|Los Angeles, CA|        2025|     -5.0|    2141|    -24.0|      0.0|         null|         null|               null|\n",
      "|2019|    1|           4|          5|               AA|            11292|              1129202|                30325|   DEN|      Denver, CO|          12892|            1289208|              32575| LAX|Los Angeles, CA|        2025|     -5.0|    2209|      4.0|      0.0|         null|         null|               null|\n",
      "|2019|    1|           5|          6|               AA|            11292|              1129202|                30325|   DEN|      Denver, CO|          12892|            1289208|              32575| LAX|Los Angeles, CA|        2025|     -4.0|    2229|     24.0|      0.0|          0.0|          0.0|                0.0|\n",
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+---------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "newFlights.filter(newFlights.ORIGIN == 'DEN' ).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's say we want to find the number of flights that originate in Denver and go to other desinations. We can combine `filter` and `groupby` on the destination city."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|      DEST_CITY_NAME|count|\n",
      "+--------------------+-----+\n",
      "|Pasco/Kennewick/R...|    2|\n",
      "|          Albany, NY|    3|\n",
      "|       Williston, ND|    5|\n",
      "|     Little Rock, AR|    5|\n",
      "|          Dayton, OH|    6|\n",
      "+--------------------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "newFlights.filter(newFlights.ORIGIN == 'DEN' )\\\n",
    "    .groupBy(newFlights.DEST_CITY_NAME)\\\n",
    "    .count().sort('count').show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We just named the column by specifying the column name (`count`) but we can also use the `col` function from the PySpark SQL library. This will also simplify setting the sort order by specific columns - for example, if we want to sort descending count but ascending on name, we would do:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import col"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|      DEST_CITY_NAME|count|\n",
      "+--------------------+-----+\n",
      "|Pasco/Kennewick/R...|    2|\n",
      "|          Albany, NY|    3|\n",
      "|     Little Rock, AR|    5|\n",
      "|       Williston, ND|    5|\n",
      "|    Bend/Redmond, OR|    6|\n",
      "|          Casper, WY|    6|\n",
      "|          Dayton, OH|    6|\n",
      "|         Buffalo, NY|    8|\n",
      "|           Greer, SC|    8|\n",
      "|Jackson/Vicksburg...|    8|\n",
      "|       Lafayette, LA|    8|\n",
      "|        Savannah, GA|    8|\n",
      "|Harlingen/San Ben...|    9|\n",
      "|       Knoxville, TN|    9|\n",
      "|      Long Beach, CA|    9|\n",
      "|      Huntsville, AL|   10|\n",
      "|      Charleston, SC|   11|\n",
      "|      Birmingham, AL|   12|\n",
      "|       Pensacola, FL|   12|\n",
      "|         Lincoln, NE|   19|\n",
      "+--------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "newFlights.filter(newFlights.ORIGIN == 'DEN' )\\\n",
    "    .groupBy(newFlights.DEST_CITY_NAME)\\\n",
    "    .count().sort(col('count').asc(), col('DEST_CITY_NAME').asc()).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Doing Joins\n",
    "\n",
    "Again, everything boils down to a join in \"big data\". We can do joins between two dataframes much as in Pandas. Let's load a second dataframe that contains airline identifiers."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "airlines = spark.read.load('unique-carriers.csv.gz',\n",
    "            format=\"csv\", sep=\",\", header=True,\n",
    "            compression=\"gzip\",\n",
    "            inferSchema=\"true\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------------+\n",
      "|Code|         Description|\n",
      "+----+--------------------+\n",
      "| 02Q|       Titan Airways|\n",
      "| 04Q|  Tradewind Aviation|\n",
      "| 05Q| Comlux Aviation, AG|\n",
      "| 06Q|Master Top Linhas...|\n",
      "| 07Q| Flair Airlines Ltd.|\n",
      "+----+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "airlines.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our flights data also has carrier information in the `OP_UNIQUE_CARRIER` column. Let's list out the distinct values by selecting that column, determining the distinct values and then showing it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+\n",
      "|OP_UNIQUE_CARRIER|\n",
      "+-----------------+\n",
      "|               UA|\n",
      "|               NK|\n",
      "|               AA|\n",
      "|               EV|\n",
      "|               B6|\n",
      "|               DL|\n",
      "|               OO|\n",
      "|               F9|\n",
      "|               YV|\n",
      "|               MQ|\n",
      "|               OH|\n",
      "|               HA|\n",
      "|               G4|\n",
      "|               YX|\n",
      "|               AS|\n",
      "|               WN|\n",
      "|               9E|\n",
      "+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "flights.select('OP_UNIQUE_CARRIER').distinct().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, let's join the airlines `Code`  with the flights `OP_UNIQUE_CARRIER`. This will result in data like the `flights` data but with two additional columns, `Code` (the join key) and `Description` (the full airline name)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+--------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+----+----+-----------------+\n",
      "|YEAR|MONTH|DAY_OF_MONTH|DAY_OF_WEEK|OP_UNIQUE_CARRIER|ORIGIN_AIRPORT_ID|ORIGIN_AIRPORT_SEQ_ID|ORIGIN_CITY_MARKET_ID|ORIGIN|ORIGIN_CITY_NAME|DEST_AIRPORT_ID|DEST_AIRPORT_SEQ_ID|DEST_CITY_MARKET_ID|DEST|DEST_CITY_NAME|CRS_DEP_TIME|DEP_DELAY|ARR_TIME|ARR_DELAY|CANCELLED|CARRIER_DELAY|WEATHER_DELAY|LATE_AIRCRAFT_DELAY|_c23|Code|      Description|\n",
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+--------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+----+----+-----------------+\n",
      "|2019|    1|          19|          6|               9E|            13487|              1348702|                31650|   MSP| Minneapolis, MN|          11193|            1119302|              33105| CVG|Cincinnati, OH|        1556|    -10.0|    1832|    -25.0|      0.0|         null|         null|               null|null|  9E|Endeavor Air Inc.|\n",
      "|2019|    1|          20|          7|               9E|            13487|              1348702|                31650|   MSP| Minneapolis, MN|          11193|            1119302|              33105| CVG|Cincinnati, OH|        1556|     -4.0|    1825|    -37.0|      0.0|         null|         null|               null|null|  9E|Endeavor Air Inc.|\n",
      "|2019|    1|          21|          1|               9E|            13487|              1348702|                31650|   MSP| Minneapolis, MN|          11193|            1119302|              33105| CVG|Cincinnati, OH|        1556|     -9.0|    1845|    -17.0|      0.0|         null|         null|               null|null|  9E|Endeavor Air Inc.|\n",
      "|2019|    1|          22|          2|               9E|            13487|              1348702|                31650|   MSP| Minneapolis, MN|          11193|            1119302|              33105| CVG|Cincinnati, OH|        1556|     -4.0|    1839|    -23.0|      0.0|         null|         null|               null|null|  9E|Endeavor Air Inc.|\n",
      "|2019|    1|          23|          3|               9E|            13487|              1348702|                31650|   MSP| Minneapolis, MN|          11193|            1119302|              33105| CVG|Cincinnati, OH|        1556|     -6.0|    1850|    -12.0|      0.0|         null|         null|               null|null|  9E|Endeavor Air Inc.|\n",
      "+----+-----+------------+-----------+-----------------+-----------------+---------------------+---------------------+------+----------------+---------------+-------------------+-------------------+----+--------------+------------+---------+--------+---------+---------+-------------+-------------+-------------------+----+----+-----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "flights.join(airlines, airlines.Code == flights.OP_UNIQUE_CARRIER).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "From here, you could *e.g.* pull out all over the Denver to Chicago flights and list them by the airline name, *etc, etc*."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Escape back into the world of RDD's"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A dataframe is composed of `Row` objects and a dataframe (and database) is just a collection of those rows. You can pull out the row objects as RDD's and then operate on those, much as we did before."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(YEAR=2019, MONTH=1, DAY_OF_MONTH=1, DAY_OF_WEEK=2, OP_UNIQUE_CARRIER='AA', ORIGIN_AIRPORT_ID=12892, ORIGIN_AIRPORT_SEQ_ID=1289208, ORIGIN_CITY_MARKET_ID=32575, ORIGIN='LAX', ORIGIN_CITY_NAME='Los Angeles, CA', DEST_AIRPORT_ID=11292, DEST_AIRPORT_SEQ_ID=1129202, DEST_CITY_MARKET_ID=30325, DEST='DEN', DEST_CITY_NAME='Denver, CO', CRS_DEP_TIME=1540, DEP_DELAY=-4.0, ARR_TIME=1915, ARR_DELAY=15.0, CANCELLED=0.0, CARRIER_DELAY=0.0, WEATHER_DELAY=0.0, LATE_AIRCRAFT_DELAY=0.0, _c23=None),\n",
       " Row(YEAR=2019, MONTH=1, DAY_OF_MONTH=2, DAY_OF_WEEK=3, OP_UNIQUE_CARRIER='AA', ORIGIN_AIRPORT_ID=12892, ORIGIN_AIRPORT_SEQ_ID=1289208, ORIGIN_CITY_MARKET_ID=32575, ORIGIN='LAX', ORIGIN_CITY_NAME='Los Angeles, CA', DEST_AIRPORT_ID=11292, DEST_AIRPORT_SEQ_ID=1129202, DEST_CITY_MARKET_ID=30325, DEST='DEN', DEST_CITY_NAME='Denver, CO', CRS_DEP_TIME=1540, DEP_DELAY=-3.0, ARR_TIME=1908, ARR_DELAY=8.0, CANCELLED=0.0, CARRIER_DELAY=None, WEATHER_DELAY=None, LATE_AIRCRAFT_DELAY=None, _c23=None),\n",
       " Row(YEAR=2019, MONTH=1, DAY_OF_MONTH=3, DAY_OF_WEEK=4, OP_UNIQUE_CARRIER='AA', ORIGIN_AIRPORT_ID=12892, ORIGIN_AIRPORT_SEQ_ID=1289208, ORIGIN_CITY_MARKET_ID=32575, ORIGIN='LAX', ORIGIN_CITY_NAME='Los Angeles, CA', DEST_AIRPORT_ID=11292, DEST_AIRPORT_SEQ_ID=1129202, DEST_CITY_MARKET_ID=30325, DEST='DEN', DEST_CITY_NAME='Denver, CO', CRS_DEP_TIME=1540, DEP_DELAY=-2.0, ARR_TIME=1904, ARR_DELAY=4.0, CANCELLED=0.0, CARRIER_DELAY=None, WEATHER_DELAY=None, LATE_AIRCRAFT_DELAY=None, _c23=None),\n",
       " Row(YEAR=2019, MONTH=1, DAY_OF_MONTH=4, DAY_OF_WEEK=5, OP_UNIQUE_CARRIER='AA', ORIGIN_AIRPORT_ID=12892, ORIGIN_AIRPORT_SEQ_ID=1289208, ORIGIN_CITY_MARKET_ID=32575, ORIGIN='LAX', ORIGIN_CITY_NAME='Los Angeles, CA', DEST_AIRPORT_ID=11292, DEST_AIRPORT_SEQ_ID=1129202, DEST_CITY_MARKET_ID=30325, DEST='DEN', DEST_CITY_NAME='Denver, CO', CRS_DEP_TIME=1540, DEP_DELAY=2.0, ARR_TIME=1855, ARR_DELAY=-5.0, CANCELLED=0.0, CARRIER_DELAY=None, WEATHER_DELAY=None, LATE_AIRCRAFT_DELAY=None, _c23=None),\n",
       " Row(YEAR=2019, MONTH=1, DAY_OF_MONTH=5, DAY_OF_WEEK=6, OP_UNIQUE_CARRIER='AA', ORIGIN_AIRPORT_ID=12892, ORIGIN_AIRPORT_SEQ_ID=1289208, ORIGIN_CITY_MARKET_ID=32575, ORIGIN='LAX', ORIGIN_CITY_NAME='Los Angeles, CA', DEST_AIRPORT_ID=11292, DEST_AIRPORT_SEQ_ID=1129202, DEST_CITY_MARKET_ID=30325, DEST='DEN', DEST_CITY_NAME='Denver, CO', CRS_DEP_TIME=1540, DEP_DELAY=22.0, ARR_TIME=1906, ARR_DELAY=6.0, CANCELLED=0.0, CARRIER_DELAY=None, WEATHER_DELAY=None, LATE_AIRCRAFT_DELAY=None, _c23=None)]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "flights.rdd.filter(lambda x: x['DEST'] == 'DEN').take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark will attempt to interpret the types of the data but it's not always successful. By default, it will use the first 100 rows to determine the types. This may fail as indicated below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "ename": "ValueError",
     "evalue": "Some of types cannot be determined by the first 100 rows, please try again with sampling",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mValueError\u001b[0m                                Traceback (most recent call last)",
      "Input \u001b[0;32mIn [50]\u001b[0m, in \u001b[0;36m<cell line: 1>\u001b[0;34m()\u001b[0m\n\u001b[0;32m----> 1\u001b[0m onlyDen \u001b[38;5;241m=\u001b[39m \u001b[43mspark\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mcreateDataFrame\u001b[49m\u001b[43m(\u001b[49m\u001b[43mflights\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mrdd\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mfilter\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;28;43;01mlambda\u001b[39;49;00m\u001b[43m \u001b[49m\u001b[43mx\u001b[49m\u001b[43m:\u001b[49m\u001b[43m \u001b[49m\u001b[43mx\u001b[49m\u001b[43m[\u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43mDEST\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[43m]\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;241;43m==\u001b[39;49m\u001b[43m \u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43mDEN\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[43m)\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m/usr/local/spark/python/pyspark/sql/session.py:894\u001b[0m, in \u001b[0;36mSparkSession.createDataFrame\u001b[0;34m(self, data, schema, samplingRatio, verifySchema)\u001b[0m\n\u001b[1;32m    889\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m has_pandas \u001b[38;5;129;01mand\u001b[39;00m \u001b[38;5;28misinstance\u001b[39m(data, pandas\u001b[38;5;241m.\u001b[39mDataFrame):\n\u001b[1;32m    890\u001b[0m     \u001b[38;5;66;03m# Create a DataFrame from pandas DataFrame.\u001b[39;00m\n\u001b[1;32m    891\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28msuper\u001b[39m(SparkSession, \u001b[38;5;28mself\u001b[39m)\u001b[38;5;241m.\u001b[39mcreateDataFrame(  \u001b[38;5;66;03m# type: ignore[call-overload]\u001b[39;00m\n\u001b[1;32m    892\u001b[0m         data, schema, samplingRatio, verifySchema\n\u001b[1;32m    893\u001b[0m     )\n\u001b[0;32m--> 894\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_create_dataframe\u001b[49m\u001b[43m(\u001b[49m\n\u001b[1;32m    895\u001b[0m \u001b[43m    \u001b[49m\u001b[43mdata\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mschema\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43msamplingRatio\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mverifySchema\u001b[49m\u001b[43m  \u001b[49m\u001b[38;5;66;43;03m# type: ignore[arg-type]\u001b[39;49;00m\n\u001b[1;32m    896\u001b[0m \u001b[43m\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m/usr/local/spark/python/pyspark/sql/session.py:934\u001b[0m, in \u001b[0;36mSparkSession._create_dataframe\u001b[0;34m(self, data, schema, samplingRatio, verifySchema)\u001b[0m\n\u001b[1;32m    931\u001b[0m         \u001b[38;5;28;01mreturn\u001b[39;00m obj\n\u001b[1;32m    933\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28misinstance\u001b[39m(data, RDD):\n\u001b[0;32m--> 934\u001b[0m     rdd, struct \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_createFromRDD\u001b[49m\u001b[43m(\u001b[49m\u001b[43mdata\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mmap\u001b[49m\u001b[43m(\u001b[49m\u001b[43mprepare\u001b[49m\u001b[43m)\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mschema\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43msamplingRatio\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m    935\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[1;32m    936\u001b[0m     rdd, struct \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_createFromLocal(\u001b[38;5;28mmap\u001b[39m(prepare, data), schema)\n",
      "File \u001b[0;32m/usr/local/spark/python/pyspark/sql/session.py:600\u001b[0m, in \u001b[0;36mSparkSession._createFromRDD\u001b[0;34m(self, rdd, schema, samplingRatio)\u001b[0m\n\u001b[1;32m    596\u001b[0m \u001b[38;5;124;03m\"\"\"\u001b[39;00m\n\u001b[1;32m    597\u001b[0m \u001b[38;5;124;03mCreate an RDD for DataFrame from an existing RDD, returns the RDD and schema.\u001b[39;00m\n\u001b[1;32m    598\u001b[0m \u001b[38;5;124;03m\"\"\"\u001b[39;00m\n\u001b[1;32m    599\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m schema \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m \u001b[38;5;129;01mor\u001b[39;00m \u001b[38;5;28misinstance\u001b[39m(schema, (\u001b[38;5;28mlist\u001b[39m, \u001b[38;5;28mtuple\u001b[39m)):\n\u001b[0;32m--> 600\u001b[0m     struct \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_inferSchema\u001b[49m\u001b[43m(\u001b[49m\u001b[43mrdd\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43msamplingRatio\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mnames\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43mschema\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m    601\u001b[0m     converter \u001b[38;5;241m=\u001b[39m _create_converter(struct)\n\u001b[1;32m    602\u001b[0m     tupled_rdd \u001b[38;5;241m=\u001b[39m rdd\u001b[38;5;241m.\u001b[39mmap(converter)\n",
      "File \u001b[0;32m/usr/local/spark/python/pyspark/sql/session.py:573\u001b[0m, in \u001b[0;36mSparkSession._inferSchema\u001b[0;34m(self, rdd, samplingRatio, names)\u001b[0m\n\u001b[1;32m    571\u001b[0m                 \u001b[38;5;28;01mbreak\u001b[39;00m\n\u001b[1;32m    572\u001b[0m         \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[0;32m--> 573\u001b[0m             \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mValueError\u001b[39;00m(\n\u001b[1;32m    574\u001b[0m                 \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mSome of types cannot be determined by the \u001b[39m\u001b[38;5;124m\"\u001b[39m\n\u001b[1;32m    575\u001b[0m                 \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mfirst 100 rows, please try again with sampling\u001b[39m\u001b[38;5;124m\"\u001b[39m\n\u001b[1;32m    576\u001b[0m             )\n\u001b[1;32m    577\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[1;32m    578\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m samplingRatio \u001b[38;5;241m<\u001b[39m \u001b[38;5;241m0.99\u001b[39m:\n",
      "\u001b[0;31mValueError\u001b[0m: Some of types cannot be determined by the first 100 rows, please try again with sampling"
     ]
    }
   ],
   "source": [
    "onlyDen = spark.createDataFrame(flights.rdd.filter(lambda x: x['DEST'] == 'DEN'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this case, the solution is to sample the data randomly -- here we're going to sample 50% of the data to determine the types:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [],
   "source": [
    "onlyDen = spark.createDataFrame(flights.rdd.filter(lambda x: x['DEST'] == 'DEN'), \n",
    "                                samplingRatio=0.5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And again, the resulting data is a `Row` type:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(YEAR=2019, MONTH=1, DAY_OF_MONTH=1, DAY_OF_WEEK=2, OP_UNIQUE_CARRIER='AA', ORIGIN_AIRPORT_ID=12892, ORIGIN_AIRPORT_SEQ_ID=1289208, ORIGIN_CITY_MARKET_ID=32575, ORIGIN='LAX', ORIGIN_CITY_NAME='Los Angeles, CA', DEST_AIRPORT_ID=11292, DEST_AIRPORT_SEQ_ID=1129202, DEST_CITY_MARKET_ID=30325, DEST='DEN', DEST_CITY_NAME='Denver, CO', CRS_DEP_TIME=1540, DEP_DELAY=-4.0, ARR_TIME=1915, ARR_DELAY=15.0, CANCELLED=0.0, CARRIER_DELAY=0.0, WEATHER_DELAY=0.0, LATE_AIRCRAFT_DELAY=0.0, _c23=None),\n",
       " Row(YEAR=2019, MONTH=1, DAY_OF_MONTH=2, DAY_OF_WEEK=3, OP_UNIQUE_CARRIER='AA', ORIGIN_AIRPORT_ID=12892, ORIGIN_AIRPORT_SEQ_ID=1289208, ORIGIN_CITY_MARKET_ID=32575, ORIGIN='LAX', ORIGIN_CITY_NAME='Los Angeles, CA', DEST_AIRPORT_ID=11292, DEST_AIRPORT_SEQ_ID=1129202, DEST_CITY_MARKET_ID=30325, DEST='DEN', DEST_CITY_NAME='Denver, CO', CRS_DEP_TIME=1540, DEP_DELAY=-3.0, ARR_TIME=1908, ARR_DELAY=8.0, CANCELLED=0.0, CARRIER_DELAY=None, WEATHER_DELAY=None, LATE_AIRCRAFT_DELAY=None, _c23=None),\n",
       " Row(YEAR=2019, MONTH=1, DAY_OF_MONTH=3, DAY_OF_WEEK=4, OP_UNIQUE_CARRIER='AA', ORIGIN_AIRPORT_ID=12892, ORIGIN_AIRPORT_SEQ_ID=1289208, ORIGIN_CITY_MARKET_ID=32575, ORIGIN='LAX', ORIGIN_CITY_NAME='Los Angeles, CA', DEST_AIRPORT_ID=11292, DEST_AIRPORT_SEQ_ID=1129202, DEST_CITY_MARKET_ID=30325, DEST='DEN', DEST_CITY_NAME='Denver, CO', CRS_DEP_TIME=1540, DEP_DELAY=-2.0, ARR_TIME=1904, ARR_DELAY=4.0, CANCELLED=0.0, CARRIER_DELAY=None, WEATHER_DELAY=None, LATE_AIRCRAFT_DELAY=None, _c23=None)]"
      ]
     },
     "execution_count": 54,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "onlyDen.take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using SQL\n",
    "\n",
    "It's clear that the Dataframe methods provide operations similar to those of SQL but in a more procedural or imperative form.\n",
    "\n",
    "PySpark also has an SQL wrapper that lets us convert a `DataFrame` into an SQL relational table."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql as sql"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [],
   "source": [
    "sqlContext = sql.SQLContext( spark.builder.getOrCreate() )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [],
   "source": [
    "sqlContext.registerDataFrameAsTable(onlyDen, \"onlyDen\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [],
   "source": [
    "sqlContext.registerDataFrameAsTable(flights, \"flights\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "From there, we can do SQL queries and a query planner will construct the series of operations needed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|   18498|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sqlContext.sql(\"SELECT COUNT(*) from onlyDEN\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|   18507|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sqlContext.sql(\"SELECT COUNT(*) from flights WHERE ORIGIN='DEN'\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|      DEST_CITY_NAME|count|\n",
      "+--------------------+-----+\n",
      "|Pasco/Kennewick/R...|    2|\n",
      "|          Albany, NY|    3|\n",
      "|     Little Rock, AR|    5|\n",
      "|       Williston, ND|    5|\n",
      "|    Bend/Redmond, OR|    6|\n",
      "|          Casper, WY|    6|\n",
      "|          Dayton, OH|    6|\n",
      "|         Buffalo, NY|    8|\n",
      "|           Greer, SC|    8|\n",
      "|Jackson/Vicksburg...|    8|\n",
      "+--------------------+-----+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sqlContext.sql(\"SELECT DEST_CITY_NAME, COUNT(*) as count from flights \\\n",
    "               WHERE ORIGIN='DEN' \\\n",
    "               GROUP BY DEST_CITY_NAME\\\n",
    "               ORDER BY count ASC, DEST_CITY_NAME ASC\").show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.10.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}