03. Improving Performance and data pipelines.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Improve data cleaning tasks by increasing performance or reducing resource requirements.\n",
    "\n",
    "### Whatiscaching?\n",
    "\n",
    "Caching in Spark:\n",
    "- Stores Data Frames inmemory or on disk\n",
    "- Improves speed on later transformations/actions\n",
    "- Reduces resource usage\n",
    "\n",
    "Disadvantages of caching\n",
    "- Very large datasets may not fit in memory\n",
    "- Local disk based caching may not be a performance improvement\n",
    "- Cached objects may not be available\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pyspark.context.SparkContext'>\n",
      "<property object at 0x000000DDA459B548>\n",
      "C:/Spark\n",
      "C:/Spark\\./bin/spark-submit\n",
      "<pyspark.sql.session.SparkSession object at 0x000000DDA58303C8>\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "import sys\n",
    "\n",
    "os.environ['SPARK_HOME'] = \"C:/Spark\"\n",
    "sys.path.append(\"C:/Spark/spark-3.1.2-bin-hadoop3.2/python/\")\n",
    "from pyspark import  SparkContext as sc# And then try to import SparkContext.\n",
    "# Verify SparkContext\n",
    "print(sc)\n",
    "\n",
    "# Print Spark version\n",
    "print(sc.version)\n",
    "#import os\n",
    "print(os.environ.get(\"SPARK_HOME\"))\n",
    "\n",
    "print(os.path.join(os.environ.get(\"SPARK_HOME\"), './bin/spark-submit'))\n",
    "\n",
    "#gateway = JavaGateway()\n",
    "\n",
    "os.environ['SPARK_HOME']=\"C:/Spark/spark-3.1.2-bin-hadoop3.2\"\n",
    "os.environ['JAVA_HOME']=\"C:/Program Files/Java/jdk1.8.0_144\"\n",
    "sys.path.append(\"C:/Spark/spark-3.1.2-bin-hadoop3.2/python\")\n",
    "os.environ['HADOOP_HOME']=\"C:/Hadoop\"\n",
    "\n",
    "\n",
    "from pyspark import SparkContext\n",
    "from pyspark import SparkConf\n",
    "\n",
    "import pyspark # only run after findspark.init()\n",
    "from pyspark.sql.functions import to_date, col\n",
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.getOrCreate()\n",
    "# Print spark\n",
    "print(spark)\n",
    "spark.conf.set(\"spark.sql.parquet.compression.codec\", \"gzip\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "processFile=('..\\\\PySpark')\n",
    "fileName =(os.listdir(processFile))\n",
    "file_path_total=[]\n",
    "def file_path(fileName):\n",
    "    #processFile=('..\\\\PySpark')\n",
    "    #fileName =(os.listdir(processFile))\n",
    "    for file in fileName:\n",
    "        if \".csv\" in file:\n",
    "            file_path = \"%s/%s\" % (processFile, file)\n",
    "            file_path_total.append(file_path)\n",
    "    return file_path_total"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Date (MM/DD/YYYY)</th>\n",
       "      <th>Flight Number</th>\n",
       "      <th>Destination Airport</th>\n",
       "      <th>Actual elapsed time (Minutes)</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>01/01/2017</td>\n",
       "      <td>0005</td>\n",
       "      <td>HNL</td>\n",
       "      <td>537</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>01/01/2017</td>\n",
       "      <td>0007</td>\n",
       "      <td>OGG</td>\n",
       "      <td>498</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>01/01/2017</td>\n",
       "      <td>0037</td>\n",
       "      <td>SFO</td>\n",
       "      <td>241</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>01/01/2017</td>\n",
       "      <td>0043</td>\n",
       "      <td>DTW</td>\n",
       "      <td>134</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>01/01/2017</td>\n",
       "      <td>0051</td>\n",
       "      <td>STL</td>\n",
       "      <td>88</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  Date (MM/DD/YYYY) Flight Number Destination Airport  \\\n",
       "0        01/01/2017          0005                 HNL   \n",
       "1        01/01/2017          0007                 OGG   \n",
       "2        01/01/2017          0037                 SFO   \n",
       "3        01/01/2017          0043                 DTW   \n",
       "4        01/01/2017          0051                 STL   \n",
       "\n",
       "  Actual elapsed time (Minutes)  \n",
       "0                           537  \n",
       "1                           498  \n",
       "2                           241  \n",
       "3                           134  \n",
       "4                            88  "
      ]
     },
     "execution_count": 65,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql import functions as F\n",
    "departures_df = spark.read.format('csv').options(Header=True).load('AA_DFW_2017_Departures_Short.csv.gz')\n",
    "\n",
    "departures_df.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Caching can improve performance when reusing DataFrames\n",
    "\n",
    "- applied the caching transformation, it doesn't take effect until an action is run. \n",
    "- action instantiates the caching after the function completes. \n",
    "- second time, there is no need to recalculate anything so it returns almost immediately."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Counting 139358 rows took 14.410116 seconds\n",
      "Counting 139358 rows again took 2.197774 seconds\n"
     ]
    }
   ],
   "source": [
    "import time\n",
    "start_time = time.time()\n",
    "\n",
    "# Add caching to the unique rows in departures_df\n",
    "departures_df = departures_df.distinct().cache()\n",
    "\n",
    "# Count the unique rows in departures_df, noting how long the operation takes\n",
    "print(\"Counting %d rows took %f seconds\" % (departures_df.count(), time.time() - start_time))\n",
    "\n",
    "# Count the rows again, noting the variance in time of a cached DataFrame\n",
    "start_time = time.time()\n",
    "print(\"Counting %d rows again took %f seconds\" % (departures_df.count(), time.time() - start_time))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Is departures_df cached?: True\n",
      "Removing departures_df from cache\n",
      "Is departures_df cached?: False\n"
     ]
    }
   ],
   "source": [
    "# Determine if departures_df is in the cache\n",
    "print(\"Is departures_df cached?: %s\" % departures_df.is_cached)\n",
    "print(\"Removing departures_df from cache\")\n",
    "\n",
    "# Remove departures_df from the cache\n",
    "departures_df.unpersist()\n",
    "\n",
    "# Check the cache status again\n",
    "print(\"Is departures_df cached?: %s\" % departures_df.is_cached)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------+-------------------+-----------------------------+\n",
      "|Date (MM/DD/YYYY)|Flight Number|Destination Airport|Actual elapsed time (Minutes)|\n",
      "+-----------------+-------------+-------------------+-----------------------------+\n",
      "|       01/01/2014|         0005|                HNL|                          519|\n",
      "|       01/01/2014|         0007|                OGG|                          505|\n",
      "|       01/01/2014|         0035|                SLC|                          174|\n",
      "|       01/01/2014|         0043|                DTW|                          153|\n",
      "|       01/01/2014|         0052|                PIT|                          137|\n",
      "|       01/01/2014|         0058|                SAN|                          174|\n",
      "|       01/01/2014|         0060|                MIA|                          155|\n",
      "|       01/01/2014|         0064|                JFK|                          185|\n",
      "|       01/01/2014|         0090|                ORD|                          126|\n",
      "|       01/01/2014|         0096|                STL|                           91|\n",
      "|       01/01/2014|         0099|                SNA|                          182|\n",
      "|       01/01/2014|         0103|                ONT|                          181|\n",
      "|       01/01/2014|         0109|                DEN|                          127|\n",
      "|       01/01/2014|         0122|                SFO|                          222|\n",
      "|       01/01/2014|         0123|                HNL|                          510|\n",
      "|       01/01/2014|         0129|                COS|                          114|\n",
      "|       01/01/2014|         0130|                DCA|                          141|\n",
      "|       01/01/2014|         0131|                SLC|                          167|\n",
      "|       01/01/2014|         0132|                STL|                           82|\n",
      "|       01/01/2014|         0140|                BWI|                          146|\n",
      "+-----------------+-------------+-------------------+-----------------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "Total rows in DataFrame 2014: \t 303756\n",
      "Time to run: 1.135980\n",
      "Total rows in all DataFrames (2014-2017):\t583718\n",
      "Time to run: 0.840558\n"
     ]
    }
   ],
   "source": [
    "# read multifile\n",
    "split_df = spark.read.csv('AA_DFW_2*_Departures_Short.csv.gz', header =True)\n",
    "df1= spark.read.csv('AA_DFW_2014_Departures_Short.csv.gz', header =True)\n",
    "df2= spark.read.csv('AA_DFW_2015_Departures_Short.csv.gz', header =True)\n",
    "split_df.show()\n",
    "start_time_b = time.time()\n",
    "print(\"Total rows in DataFrame 2014: \\t %d\" %(df1.count()+df2.count()))\n",
    "print(\"Time to run: %f\" % (time.time() - start_time_b))\n",
    "start_time_b = time.time()\n",
    "print(\"Total rows in all DataFrames (2014-2017):\\t%d\" % split_df.count())\n",
    "print(\"Time to run: %f\" % (time.time() - start_time_b))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Cluster configurations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Name: pyspark-shell\n",
      "Driver TCP port: 56657\n",
      "Number of partitions: 500\n"
     ]
    }
   ],
   "source": [
    "# Name of the Spark application instance\n",
    "app_name = spark.conf.get('spark.app.name')\n",
    "\n",
    "# Driver TCP port\n",
    "driver_tcp_port = spark.conf.get('spark.driver.port')\n",
    "\n",
    "# Number of join partitions\n",
    "num_partitions = spark.conf.get('spark.sql.shuffle.partitions')\n",
    "\n",
    "# Show the results\n",
    "print(\"Name: %s\" % app_name)\n",
    "print(\"Driver TCP port: %s\" % driver_tcp_port)\n",
    "print(\"Number of partitions: %s\" % num_partitions)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Writing Spark configurations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Partition count before change: 200\n",
      "Partition count after change: 500\n"
     ]
    }
   ],
   "source": [
    "# Store the number of partitions in variable\n",
    "before = departures_df.rdd.getNumPartitions()\n",
    "\n",
    "# Configure Spark to use 500 partitions\n",
    "spark.conf.set('spark.sql.shuffle.partitions', 500)\n",
    "\n",
    "# Recreate the DataFrame using the departures data file\n",
    "departures_df = spark.read.csv('AA_DFW_2014_Departures_Short.csv.gz').distinct()\n",
    "\n",
    "# Print the number of partitions for each instance\n",
    "print(\"Partition count before change: %d\" % before)\n",
    "print(\"Partition count after change: %d\" % departures_df.rdd.getNumPartitions())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### What is shuffling?\n",
    "- Shufling refers to moving data around to various workers to complete a task \n",
    "- Hides complexity from the user \n",
    "- Can be slow to complete \n",
    "- Lowers overall throughput\n",
    "- Is often necessary,but try to minimize\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "*(2) HashAggregate(keys=[_c0#1014, _c1#1015, _c2#1016, _c3#1017], functions=[])\n",
      "+- Exchange hashpartitioning(_c0#1014, _c1#1015, _c2#1016, _c3#1017, 500), ENSURE_REQUIREMENTS, [id=#752]\n",
      "   +- *(1) HashAggregate(keys=[_c0#1014, _c1#1015, _c2#1016, _c3#1017], functions=[])\n",
      "      +- FileScan csv [_c0#1014,_c1#1015,_c2#1016,_c3#1017] Batched: false, DataFilters: [], Format: CSV, Location: InMemoryFileIndex[file:/C:/Users/lienphuong/Documents/GitHub/PySpark/AA_DFW_2014_Departures_Short..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<_c0:string,_c1:string,_c2:string,_c3:string>\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "departures_df.explain()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "*(3) HashAggregate(keys=[_c2#1016], functions=[])\n",
      "+- Exchange hashpartitioning(_c2#1016, 500), ENSURE_REQUIREMENTS, [id=#811]\n",
      "   +- *(2) HashAggregate(keys=[_c2#1016], functions=[])\n",
      "      +- *(2) HashAggregate(keys=[_c0#1014, _c1#1015, _c2#1016, _c3#1017], functions=[])\n",
      "         +- Exchange hashpartitioning(_c0#1014, _c1#1015, _c2#1016, _c3#1017, 500), ENSURE_REQUIREMENTS, [id=#806]\n",
      "            +- *(1) HashAggregate(keys=[_c0#1014, _c1#1015, _c2#1016, _c3#1017], functions=[])\n",
      "               +- FileScan csv [_c0#1014,_c1#1015,_c2#1016,_c3#1017] Batched: false, DataFilters: [], Format: CSV, Location: InMemoryFileIndex[file:/C:/Users/lienphuong/Documents/GitHub/PySpark/AA_DFW_2014_Departures_Short..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<_c0:string,_c1:string,_c2:string,_c3:string>\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "departures_df_1=departures_df.select(departures_df['_c2']).distinct()\n",
    "departures_df_1.explain()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "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",
      "+----+-----+---+--------+---------+--------+---------+-------+-------+------+------+----+--------+--------+----+------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+---+--------------------+----------+-----------+----+---+---+\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",
      "+---+--------------------+----------+-----------+----+---+---+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "flights_df= spark.read.csv(\"flights_small.csv\", header =True)\n",
    "flights_df.show(5)\n",
    "airports_df= spark.read.csv(\"airports.csv\", header =True)\n",
    "airports_df.show(5)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1397"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "airports_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>year</th>\n",
       "      <th>month</th>\n",
       "      <th>day</th>\n",
       "      <th>dep_time</th>\n",
       "      <th>dep_delay</th>\n",
       "      <th>arr_time</th>\n",
       "      <th>arr_delay</th>\n",
       "      <th>carrier</th>\n",
       "      <th>tailnum</th>\n",
       "      <th>flight</th>\n",
       "      <th>...</th>\n",
       "      <th>distance</th>\n",
       "      <th>hour</th>\n",
       "      <th>minute</th>\n",
       "      <th>faa</th>\n",
       "      <th>name</th>\n",
       "      <th>lat</th>\n",
       "      <th>lon</th>\n",
       "      <th>alt</th>\n",
       "      <th>tz</th>\n",
       "      <th>dst</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2014</td>\n",
       "      <td>12</td>\n",
       "      <td>8</td>\n",
       "      <td>658</td>\n",
       "      <td>-7</td>\n",
       "      <td>935</td>\n",
       "      <td>-5</td>\n",
       "      <td>VX</td>\n",
       "      <td>N846VA</td>\n",
       "      <td>1780</td>\n",
       "      <td>...</td>\n",
       "      <td>954</td>\n",
       "      <td>6</td>\n",
       "      <td>58</td>\n",
       "      <td>LAX</td>\n",
       "      <td>Los Angeles Intl</td>\n",
       "      <td>33.942536</td>\n",
       "      <td>-118.408075</td>\n",
       "      <td>126</td>\n",
       "      <td>-8</td>\n",
       "      <td>A</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2014</td>\n",
       "      <td>1</td>\n",
       "      <td>22</td>\n",
       "      <td>1040</td>\n",
       "      <td>5</td>\n",
       "      <td>1505</td>\n",
       "      <td>5</td>\n",
       "      <td>AS</td>\n",
       "      <td>N559AS</td>\n",
       "      <td>851</td>\n",
       "      <td>...</td>\n",
       "      <td>2677</td>\n",
       "      <td>10</td>\n",
       "      <td>40</td>\n",
       "      <td>HNL</td>\n",
       "      <td>Honolulu Intl</td>\n",
       "      <td>21.318681</td>\n",
       "      <td>-157.922428</td>\n",
       "      <td>13</td>\n",
       "      <td>-10</td>\n",
       "      <td>N</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2014</td>\n",
       "      <td>3</td>\n",
       "      <td>9</td>\n",
       "      <td>1443</td>\n",
       "      <td>-2</td>\n",
       "      <td>1652</td>\n",
       "      <td>2</td>\n",
       "      <td>VX</td>\n",
       "      <td>N847VA</td>\n",
       "      <td>755</td>\n",
       "      <td>...</td>\n",
       "      <td>679</td>\n",
       "      <td>14</td>\n",
       "      <td>43</td>\n",
       "      <td>SFO</td>\n",
       "      <td>San Francisco Intl</td>\n",
       "      <td>37.618972</td>\n",
       "      <td>-122.374889</td>\n",
       "      <td>13</td>\n",
       "      <td>-8</td>\n",
       "      <td>A</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2014</td>\n",
       "      <td>4</td>\n",
       "      <td>9</td>\n",
       "      <td>1705</td>\n",
       "      <td>45</td>\n",
       "      <td>1839</td>\n",
       "      <td>34</td>\n",
       "      <td>WN</td>\n",
       "      <td>N360SW</td>\n",
       "      <td>344</td>\n",
       "      <td>...</td>\n",
       "      <td>569</td>\n",
       "      <td>17</td>\n",
       "      <td>5</td>\n",
       "      <td>SJC</td>\n",
       "      <td>Norman Y Mineta San Jose Intl</td>\n",
       "      <td>37.3626</td>\n",
       "      <td>-121.929022</td>\n",
       "      <td>62</td>\n",
       "      <td>-8</td>\n",
       "      <td>A</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>2014</td>\n",
       "      <td>3</td>\n",
       "      <td>9</td>\n",
       "      <td>754</td>\n",
       "      <td>-1</td>\n",
       "      <td>1015</td>\n",
       "      <td>1</td>\n",
       "      <td>AS</td>\n",
       "      <td>N612AS</td>\n",
       "      <td>522</td>\n",
       "      <td>...</td>\n",
       "      <td>937</td>\n",
       "      <td>7</td>\n",
       "      <td>54</td>\n",
       "      <td>BUR</td>\n",
       "      <td>Bob Hope</td>\n",
       "      <td>34.200667</td>\n",
       "      <td>-118.358667</td>\n",
       "      <td>778</td>\n",
       "      <td>-8</td>\n",
       "      <td>A</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 23 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "   year month day dep_time dep_delay arr_time arr_delay carrier tailnum  \\\n",
       "0  2014    12   8      658        -7      935        -5      VX  N846VA   \n",
       "1  2014     1  22     1040         5     1505         5      AS  N559AS   \n",
       "2  2014     3   9     1443        -2     1652         2      VX  N847VA   \n",
       "3  2014     4   9     1705        45     1839        34      WN  N360SW   \n",
       "4  2014     3   9      754        -1     1015         1      AS  N612AS   \n",
       "\n",
       "  flight  ... distance hour minute  faa                           name  \\\n",
       "0   1780  ...      954    6     58  LAX               Los Angeles Intl   \n",
       "1    851  ...     2677   10     40  HNL                  Honolulu Intl   \n",
       "2    755  ...      679   14     43  SFO             San Francisco Intl   \n",
       "3    344  ...      569   17      5  SJC  Norman Y Mineta San Jose Intl   \n",
       "4    522  ...      937    7     54  BUR                       Bob Hope   \n",
       "\n",
       "         lat          lon  alt   tz dst  \n",
       "0  33.942536  -118.408075  126   -8   A  \n",
       "1  21.318681  -157.922428   13  -10   N  \n",
       "2  37.618972  -122.374889   13   -8   A  \n",
       "3    37.3626  -121.929022   62   -8   A  \n",
       "4  34.200667  -118.358667  778   -8   A  \n",
       "\n",
       "[5 rows x 23 columns]"
      ]
     },
     "execution_count": 54,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.functions import *\n",
    "# Join the flights_df and aiports_df DataFrames\n",
    "normal_df = flights_df.join(airports_df,flights_df[\"dest\"] == airports_df[\"faa\"] )\n",
    "\n",
    "# Show the query plan\n",
    "normal_df.limit(5).toPandas()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "*(2) BroadcastHashJoin [dest#1077], [faa#1195], Inner, BuildRight, false\n",
      ":- *(2) Filter isnotnull(dest#1077)\n",
      ":  +- FileScan csv [year#1066,month#1067,day#1068,dep_time#1069,dep_delay#1070,arr_time#1071,arr_delay#1072,carrier#1073,tailnum#1074,flight#1075,origin#1076,dest#1077,air_time#1078,distance#1079,hour#1080,minute#1081] Batched: false, DataFilters: [isnotnull(dest#1077)], Format: CSV, Location: InMemoryFileIndex[file:/C:/Users/lienphuong/Documents/GitHub/PySpark/flights_small.csv], PartitionFilters: [], PushedFilters: [IsNotNull(dest)], ReadSchema: struct<year:string,month:string,day:string,dep_time:string,dep_delay:string,arr_time:string,arr_d...\n",
      "+- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [id=#1158]\n",
      "   +- *(1) Filter isnotnull(faa#1195)\n",
      "      +- FileScan csv [faa#1195,name#1196,lat#1197,lon#1198,alt#1199,tz#1200,dst#1201] Batched: false, DataFilters: [isnotnull(faa#1195)], Format: CSV, Location: InMemoryFileIndex[file:/C:/Users/lienphuong/Documents/GitHub/PySpark/airports.csv], PartitionFilters: [], PushedFilters: [IsNotNull(faa)], ReadSchema: struct<faa:string,name:string,lat:string,lon:string,alt:string,tz:string,dst:string>\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "normal_df.explain()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>year</th>\n",
       "      <th>month</th>\n",
       "      <th>day</th>\n",
       "      <th>dep_time</th>\n",
       "      <th>dep_delay</th>\n",
       "      <th>arr_time</th>\n",
       "      <th>arr_delay</th>\n",
       "      <th>carrier</th>\n",
       "      <th>tailnum</th>\n",
       "      <th>flight</th>\n",
       "      <th>...</th>\n",
       "      <th>distance</th>\n",
       "      <th>hour</th>\n",
       "      <th>minute</th>\n",
       "      <th>faa</th>\n",
       "      <th>name</th>\n",
       "      <th>lat</th>\n",
       "      <th>lon</th>\n",
       "      <th>alt</th>\n",
       "      <th>tz</th>\n",
       "      <th>dst</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2014</td>\n",
       "      <td>12</td>\n",
       "      <td>8</td>\n",
       "      <td>658</td>\n",
       "      <td>-7</td>\n",
       "      <td>935</td>\n",
       "      <td>-5</td>\n",
       "      <td>VX</td>\n",
       "      <td>N846VA</td>\n",
       "      <td>1780</td>\n",
       "      <td>...</td>\n",
       "      <td>954</td>\n",
       "      <td>6</td>\n",
       "      <td>58</td>\n",
       "      <td>LAX</td>\n",
       "      <td>Los Angeles Intl</td>\n",
       "      <td>33.942536</td>\n",
       "      <td>-118.408075</td>\n",
       "      <td>126</td>\n",
       "      <td>-8</td>\n",
       "      <td>A</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2014</td>\n",
       "      <td>1</td>\n",
       "      <td>22</td>\n",
       "      <td>1040</td>\n",
       "      <td>5</td>\n",
       "      <td>1505</td>\n",
       "      <td>5</td>\n",
       "      <td>AS</td>\n",
       "      <td>N559AS</td>\n",
       "      <td>851</td>\n",
       "      <td>...</td>\n",
       "      <td>2677</td>\n",
       "      <td>10</td>\n",
       "      <td>40</td>\n",
       "      <td>HNL</td>\n",
       "      <td>Honolulu Intl</td>\n",
       "      <td>21.318681</td>\n",
       "      <td>-157.922428</td>\n",
       "      <td>13</td>\n",
       "      <td>-10</td>\n",
       "      <td>N</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2014</td>\n",
       "      <td>3</td>\n",
       "      <td>9</td>\n",
       "      <td>1443</td>\n",
       "      <td>-2</td>\n",
       "      <td>1652</td>\n",
       "      <td>2</td>\n",
       "      <td>VX</td>\n",
       "      <td>N847VA</td>\n",
       "      <td>755</td>\n",
       "      <td>...</td>\n",
       "      <td>679</td>\n",
       "      <td>14</td>\n",
       "      <td>43</td>\n",
       "      <td>SFO</td>\n",
       "      <td>San Francisco Intl</td>\n",
       "      <td>37.618972</td>\n",
       "      <td>-122.374889</td>\n",
       "      <td>13</td>\n",
       "      <td>-8</td>\n",
       "      <td>A</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2014</td>\n",
       "      <td>4</td>\n",
       "      <td>9</td>\n",
       "      <td>1705</td>\n",
       "      <td>45</td>\n",
       "      <td>1839</td>\n",
       "      <td>34</td>\n",
       "      <td>WN</td>\n",
       "      <td>N360SW</td>\n",
       "      <td>344</td>\n",
       "      <td>...</td>\n",
       "      <td>569</td>\n",
       "      <td>17</td>\n",
       "      <td>5</td>\n",
       "      <td>SJC</td>\n",
       "      <td>Norman Y Mineta San Jose Intl</td>\n",
       "      <td>37.3626</td>\n",
       "      <td>-121.929022</td>\n",
       "      <td>62</td>\n",
       "      <td>-8</td>\n",
       "      <td>A</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>2014</td>\n",
       "      <td>3</td>\n",
       "      <td>9</td>\n",
       "      <td>754</td>\n",
       "      <td>-1</td>\n",
       "      <td>1015</td>\n",
       "      <td>1</td>\n",
       "      <td>AS</td>\n",
       "      <td>N612AS</td>\n",
       "      <td>522</td>\n",
       "      <td>...</td>\n",
       "      <td>937</td>\n",
       "      <td>7</td>\n",
       "      <td>54</td>\n",
       "      <td>BUR</td>\n",
       "      <td>Bob Hope</td>\n",
       "      <td>34.200667</td>\n",
       "      <td>-118.358667</td>\n",
       "      <td>778</td>\n",
       "      <td>-8</td>\n",
       "      <td>A</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 23 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "   year month day dep_time dep_delay arr_time arr_delay carrier tailnum  \\\n",
       "0  2014    12   8      658        -7      935        -5      VX  N846VA   \n",
       "1  2014     1  22     1040         5     1505         5      AS  N559AS   \n",
       "2  2014     3   9     1443        -2     1652         2      VX  N847VA   \n",
       "3  2014     4   9     1705        45     1839        34      WN  N360SW   \n",
       "4  2014     3   9      754        -1     1015         1      AS  N612AS   \n",
       "\n",
       "  flight  ... distance hour minute  faa                           name  \\\n",
       "0   1780  ...      954    6     58  LAX               Los Angeles Intl   \n",
       "1    851  ...     2677   10     40  HNL                  Honolulu Intl   \n",
       "2    755  ...      679   14     43  SFO             San Francisco Intl   \n",
       "3    344  ...      569   17      5  SJC  Norman Y Mineta San Jose Intl   \n",
       "4    522  ...      937    7     54  BUR                       Bob Hope   \n",
       "\n",
       "         lat          lon  alt   tz dst  \n",
       "0  33.942536  -118.408075  126   -8   A  \n",
       "1  21.318681  -157.922428   13  -10   N  \n",
       "2  37.618972  -122.374889   13   -8   A  \n",
       "3    37.3626  -121.929022   62   -8   A  \n",
       "4  34.200667  -118.358667  778   -8   A  \n",
       "\n",
       "[5 rows x 23 columns]"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Import the broadcast method from pyspark.sql.functions\n",
    "from pyspark.sql.functions import broadcast\n",
    "\n",
    "# Join the flights_df and airports_df DataFrames using broadcasting\n",
    "broadcast_df = flights_df.join(broadcast(airports_df), \\\n",
    "    flights_df[\"dest\"] == airports_df[\"faa\"] )\n",
    "\n",
    "broadcast_df.limit(5).toPandas()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "*(2) BroadcastHashJoin [dest#1077], [faa#1195], Inner, BuildRight, false\n",
      ":- *(2) Filter isnotnull(dest#1077)\n",
      ":  +- FileScan csv [year#1066,month#1067,day#1068,dep_time#1069,dep_delay#1070,arr_time#1071,arr_delay#1072,carrier#1073,tailnum#1074,flight#1075,origin#1076,dest#1077,air_time#1078,distance#1079,hour#1080,minute#1081] Batched: false, DataFilters: [isnotnull(dest#1077)], Format: CSV, Location: InMemoryFileIndex[file:/C:/Users/lienphuong/Documents/GitHub/PySpark/flights_small.csv], PartitionFilters: [], PushedFilters: [IsNotNull(dest)], ReadSchema: struct<year:string,month:string,day:string,dep_time:string,dep_delay:string,arr_time:string,arr_d...\n",
      "+- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [id=#1325]\n",
      "   +- *(1) Filter isnotnull(faa#1195)\n",
      "      +- FileScan csv [faa#1195,name#1196,lat#1197,lon#1198,alt#1199,tz#1200,dst#1201] Batched: false, DataFilters: [isnotnull(faa#1195)], Format: CSV, Location: InMemoryFileIndex[file:/C:/Users/lienphuong/Documents/GitHub/PySpark/airports.csv], PartitionFilters: [], PushedFilters: [IsNotNull(faa)], ReadSchema: struct<faa:string,name:string,lat:string,lon:string,alt:string,tz:string,dst:string>\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Show the query plan and compare against the original\n",
    "broadcast_df.explain()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Normal count:\t\t10000\tduration: 1.359908\n",
      "Broadcast count:\t10000\tduration: 0.620395\n"
     ]
    }
   ],
   "source": [
    "start_time = time.time()\n",
    "# Count the number of rows in the normal DataFrame\n",
    "normal_count = normal_df.count()\n",
    "normal_duration = time.time() - start_time\n",
    "\n",
    "start_time = time.time()\n",
    "# Count the number of rows in the broadcast DataFrame\n",
    "broadcast_count = broadcast_df.count()\n",
    "broadcast_duration = time.time() - start_time\n",
    "\n",
    "# Print the counts and the duration of the tests\n",
    "print(\"Normal count:\\t\\t%d\\tduration: %f\" % (normal_count, normal_duration))\n",
    "print(\"Broadcast count:\\t%d\\tduration: %f\" % (broadcast_count, broadcast_duration))"
   ]
  }
 ],
 "metadata": {
  "hide_input": false,
  "kernelspec": {
   "display_name": "Python 3.7.7 64-bit",
   "language": "python",
   "name": "python37764bit31c54662e790487c9a6cdc87294e2b12"
  },
  "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.7.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}