<jupyter_start><jupyter_text>Yearly number of carriers' flights provided for each origin-dest pair (route)In this notebook we compute the yearly number of flights provided by the carriers for each route.In the other notebooks other aspects of the dataset were analysed but no information about the carriers were inspected.\begin{equation}CountOfFlights(carrier, year) = \sum_{f\space in \space flights(year)} x_f\, , \,\,\, x_f = \begin{cases} 1, & \text{if} \space \space f[Carrier] == carrier \\ 0, & \text{otherwise} \end{cases}\end{equation}  Execution modeIf development mode is turned on, all the calculations will be performed on a reduced instance of the data set.This allows to *save computation time* while developing.<jupyter_code># development mode
# False when performing real analytics
DEV = False

### threads to be used to run spark worker nodes locally
spark_local_threads = 4

# default transformed dataset path
transformed_dataset_path = '../dataset/05_cache_yearly_carries_flights_count_for_each_route.parquet'
missings_dataframe_path = '../dataset/05_cache_yearly_carries_flights_count_for_each_route_missings.pkl'
origins_dests_path = '../dataset/05_cache_origins_dests.pkl'<jupyter_output><empty_output><jupyter_text>Find and import Apache Spark on current machine<jupyter_code>import findspark
findspark.init()
# imports
import pickle
from pathlib import Path

from pyspark.sql import SparkSession
import pyspark.sql.functions as F
from pyspark.sql.types import *

import numpy as np
import pandas as pd

from missing_values_utils import *<jupyter_output><empty_output><jupyter_text>Create a new session<jupyter_code># Build a Spark SQL Session for DataFrames
master = 'local[{}]'.format(spark_local_threads)
appName = 'Yearly carriers filghts count for each route'
spark = SparkSession  \
    .builder          \
    .appName(appName) \
    .master(master)   \
    .getOrCreate()<jupyter_output><empty_output><jupyter_text>Load dataIf there exists a cached version of the already transformed dataset, just load it and skip all the unnecessary computation.<jupyter_code>do_computation = True

if not DEV and Path(transformed_dataset_path).is_dir():
    df = spark.read.load(transformed_dataset_path)
    do_computation = False<jupyter_output><empty_output><jupyter_text>Try to load the optimized parquet format data set. If parquet data set is not found, load full compressed data sets, reduce and save them.<jupyter_code>from preprocessing_utils import *

if do_computation:
    if DEV:
        # DEV preprocessing
        perform_DEV_dataset_preprocessing(spark)
    else:
        # Production preprocessing
        perform_dataset_preprocessing(spark)
# Load the parquet dataset
if do_computation:
    if DEV:
        # Load DEV dataset
        df = load_DEV_preprocessed_dataset(spark)
    else:
        # Load production dataset
        df = load_preprocessed_dataset(spark)<jupyter_output>Peprocessed dataset loaded.
../dataset/preprocessed_dataset.parquet<jupyter_text>Attributes selection<jupyter_code>if do_computation:
    # Keep only the dimensions we need
    df = df.select('Year', 'Cancelled' ,'Origin', 'Dest', 'UniqueCarrier')
    # Explore the schema
    df.printSchema()
# get years from dataset
if do_computation:
    df_years = df.select('Year').distinct().toPandas()['Year']
else:
    df_years = range(1994,2009)

df_years = sorted(df_years)
print(df_years)<jupyter_output>[1994, 1995, 1996, 1997, 1998, 1999, 2000, 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008]<jupyter_text>Missing values exploration and exclusion<jupyter_code>df.count()
df, df_missing = get_df_missing_and_drop_all_null(df, do_computation, missings_dataframe_path)
df.count()<jupyter_output><empty_output><jupyter_text>Compute yearly carriers filghts count per route<jupyter_code># Drop cancelled flights
if do_computation:
    df = df.drop(df['Cancelled'] == 1)
    df = df.drop('Cancelled')
if do_computation:
    df = df.withColumn("Count", F.lit(1))

    df = df.groupBy(['Year', 'Origin', 'Dest', 'UniqueCarrier'])  \
           .sum('Count')                                          \
           .withColumnRenamed('sum(Count)', 'Count')         

    df.show(10)
if do_computation:
    df.write.mode('overwrite').save(transformed_dataset_path, format='parquet')
df = spark.read.load(transformed_dataset_path)<jupyter_output><empty_output><jupyter_text>Data format optimazion<jupyter_code>#We create a dictionary where for each origin there are associated all the possible destinations
def get_origins_dests_dict(df):
    print('Started. Please wait...')
    from datetime import datetime
    begin = datetime.now()
    
    origin_dests = {}

    origins = df.select('Origin')     \
                .orderBy('Origin')    \
                .distinct()           \
                .rdd                  \
                .map(lambda x : x[0]) \
                .collect()

    for origin in origins:
        dests = df.filter(df.Origin == origin) \
                  .select('Dest')              \
                  .orderBy('Dest')             \
                  .distinct()                  \
                  .rdd                         \
                  .map(lambda x : x[0])        \
                  .collect()    
        origin_dests[origin] = dests 
    
    end = datetime.now()
    print('Complete. Elapsed time: {}'.format(end - begin))
        
    return origin_dests
if Path(origins_dests_path).is_file() and not do_computation:
    # load from cache
    with open(r'' + origins_dests_path, 'rb') as pf:
        origins_dests = pickle.load(pf)
else:
    # compute
    origins_dests = get_origins_dests_dict(df)

    # cache it
    with open(r'' + origins_dests_path, 'wb') as pf:
        pickle.dump(origins_dests, pf)
# Carriers
unique_carriers = df.select('UniqueCarrier').distinct().toPandas()['UniqueCarrier']
unique_carriers = sorted(unique_carriers)
unique_carriers
# Airports
airports_origin = df.select('Origin').distinct().orderBy('Origin').rdd.map(lambda x : x[0]).collect()
airports_dest = df.select('Dest').distinct().orderBy('Dest').rdd.map(lambda x : x[0]).collect()<jupyter_output><empty_output><jupyter_text>Data visualization functionsScroll down to see the charts<jupyter_code># Hide warnings if there are any
import warnings
warnings.filterwarnings('ignore')

%matplotlib inline


from ipywidgets import interact_manual
import ipywidgets as widgets
import matplotlib.pyplot as plt
from matplotlib.font_manager import FontProperties
def get_count_df(airport_origin, airport_dest, years, df):
    df_selection = df.filter(df.Year.isin(*years))        \
                     .filter(df.Origin == airport_origin) \
                     .filter(df.Dest == airport_dest)     \
                     .select('UniqueCarrier', 'Count' )   \
                     .toPandas()
    
    # create dictionary: key-> carrier, value-> 0
    data = dict.fromkeys(unique_carriers, 0)
    for idx, r in df_selection.iterrows():
        data[r['UniqueCarrier']] += r['Count']
        
    return pd.DataFrame.from_dict(data, orient='index')


def plot_count_carriers(airport_origin, airport_dest, years, df, ax):
    df = get_count_df(airport_origin, airport_dest, years, df)
    title = 'Unique Carrier count for Origin-Dest and Year'
    df.plot.bar(title=title, rot=90, ax=ax, legend=False)


def on_change(change):
    # exclude the origin ariport and all the aiports with count 0 for all the carriers
    global airports2_w
    global df
    airports_dx = origins_dests[change['new']]   
    airports2_w.options = airports_dx
    airports2_w.value = 'N/A' if not airports_dx else airports_dx[0]
def ui_callback(airport_origin, airport_dest, years, df):
    plt.figure(figsize=(12,15))
    plt.clf()
    ax = plt.subplot(211)
    plot_count_carriers(airport_origin, airport_dest,range(years[0], years[1] + 1), df, ax)
    
    plt.subplots_adjust(top=0.92, bottom=0.08, left=0.10, right=0.95, hspace=0.25, wspace=0.35)
    plt.show()


# Years selection range
years = [(str(y), y) for y in df_years]
years_w = widgets.SelectionRangeSlider(options=years,
                                       index=(0, 2),
                                       description='Years',
                                       continuous_update=False)
# Airports selection menus
airports1_w = widgets.Dropdown(options     = airports_origin,
                               value       = airports_origin[0],
                               description = 'Airport_Origin')
# set the listener function to the observer
airports1_w.observe(on_change, 'value')

airports2_w = widgets.Dropdown(options     = origins_dests['ABE'],
                               value       = origins_dests['ABE'][0],
                               description = 'Airport_Dest')
out = widgets.interactive_output(ui_callback, {'airport_origin': airports1_w,'airport_dest': airports2_w, 'years': years_w, 'df': widgets.fixed(df)})
ui = widgets.HBox([airports1_w, airports2_w, years_w])<jupyter_output><empty_output><jupyter_text>Data VisualizationAnalytics showhing yearly number of flights provided by the carriers for each route are reported below.  Missing values and valid data<jupyter_code>plot_missing_values_stacked_bar(df_missing)<jupyter_output><empty_output><jupyter_text>Yearly number of flights provided by the carriers for each routeFor the user convenience, the Aiport_Dest dropdown is automatically filled with only the airports that are valid destinations with respect to the selected Airport_Origin. Change the Airport_Origin value to update the Aiport_Dest dropdown.> **WARNING**>>If the resulting chart is empty, it means that for the selected years that are no flights. In this case, by selecting all the available years, at least one bar will be visualized. Data discoveriesWe decided to keep the absolute value of flights managed by each carrier and not to show the percentage w.r.t. the totality of flights on the route because we noticed a very high variance of the possible results. Some routes are much more covered than others. Suggested tests- ABE-AZO: 1 flight only (2006)- ABE-ATL interesting before/after 2000- DEN-ABE- DEN-COS- BOS-JFKList of airports in the United States: https://en.wikipedia.org/wiki/List_of_airports_in_the_United_StatesList of airline codes: https://en.wikipedia.org/wiki/List_of_airline_codes<jupyter_code>display(ui, out)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09624074608210749, 'fraction_numerical': 0.013652533410249015, 'mean_word_length': 3.4759036144578315, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 52, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '410630', 'license': 'MIT', 'n_tokens_mistral': 3314, 'n_tokens_neox': 3093, 'n_words': 843, 'path': 'notebooks/05-yearly-carriers-flights-count-for-each-route.ipynb', 'repo_name': 'polimi-mt-acg/USA-airlines-statistics-1994-2008', 'size': '10401'}