<jupyter_start><jupyter_text>Imports<jupyter_code>from os import environ, path
environ["SPARK_HOME"] = "/home/students/spark-2.2.0"

import findspark
findspark.init()

from io import StringIO
import csv
import matplotlib.pyplot as plt
%matplotlib inline

from datetime import datetime
from collections import namedtuple
from operator import add, itemgetter
from pyspark import SparkContext<jupyter_output><empty_output><jupyter_text>Get Some Context<jupyter_code># Create a Spark context to use
sc = SparkContext(appName="Flight Delay Analysis")<jupyter_output><empty_output><jupyter_text>Load the Data<jupyter_code># Module Constants
AIRLINES_FILE = "/home/students/data/ontime/airlines.csv"
FLIGHTS_FILE = "/home/students/data/ontime/flights.csv"

DATE_FMT = "%Y-%m-%d"
TIME_FMT = "%H%M"

fields   = ('date', 'airline', 'flightnum', 'origin', 'dest', 'dep',
            'dep_delay', 'arv', 'arv_delay', 'airtime', 'distance')

Flight   = namedtuple('Flight', fields)
def split(line):
    """
    Operator function for splitting a line with csv module
    """
    reader = csv.reader(StringIO(line))
    return next(reader)
def parse(row):
    """
    Parses a row and returns a named tuple.
    """

    row[0]  = datetime.strptime(row[0], DATE_FMT).date()
    row[5]  = datetime.strptime(row[5], TIME_FMT).time()
    row[6]  = float(row[6])
    row[7]  = datetime.strptime(row[7], TIME_FMT).time()
    row[8]  = float(row[8])
    row[9]  = float(row[9])
    row[10] = float(row[10])
    return Flight(*row[:11])
# Load the airlines lookup dictionary
airlines = dict(sc.textFile(AIRLINES_FILE).map(split).collect())
airlines
# Broadcast the lookup dictionary to the cluster
airline_lookup = sc.broadcast(airlines)
# Read the CSV data into an RDD
flights = sc.textFile(FLIGHTS_FILE).map(split).map(parse)
flights.take(1)
# Map the total delay to the airline (joined using the broadcast value)
delays  = flights.map(lambda f: (airline_lookup.value[f.airline],
                                 add(f.dep_delay, f.arv_delay)))
delays.take(1)
# Reduce the total delay for the month to the airline
delays  = delays.reduceByKey(add).collect()
delays  = sorted(delays, key=itemgetter(1))
delays
# Provide output from the driver
for d in delays:
    print("{} minutes delayed \t {}".format(d[1], d[0]))
def plot(delays):
    """
    Show a bar chart of the total delay per airline
    """
    airlines = [d[0] for d in delays]
    minutes  = [d[1] for d in delays]
    index    = list(range(len(airlines)))

    fig, axe = plt.subplots()
    bars = axe.barh(index, minutes)

    # Add the total minutes to the right
    for idx, air, min in zip(index, airlines, minutes):
        if min > 0:
            bars[idx].set_color('#d9230f')
            axe.annotate(" %0.0f min" % min, xy=(min+1, idx+0.5), va='center')
        else:
            bars[idx].set_color('#469408')
            axe.annotate(" %0.0f min" % min, xy=(10, idx+0.5), va='center')

    # Set the ticks
    ticks = plt.yticks([idx+ 0.5 for idx in index], airlines)
    xt = plt.xticks()[0]
    plt.xticks(xt, [' '] * len(xt))

    # Minimize chart junk
    plt.grid(axis = 'x', color ='white', linestyle='-')

    plt.title('Total Minutes Delayed per Airline')
    plt.show()
# Show a bar chart of the delays
plot(delays)<jupyter_output><empty_output><jupyter_text>Shut it down<jupyter_code>sc.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.14138134592680046, 'fraction_numerical': 0.01564344746162928, 'mean_word_length': 3.8276353276353277, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '541106', 'license': 'Apache-2.0', 'n_tokens_mistral': 1236, 'n_tokens_neox': 1195, 'n_words': 305, 'path': 'code/day_2/1 - Flight Delays.ipynb', 'repo_name': 'anjijava16/data-analytics-machine-learning-big-data', 'size': '3388'}