<jupyter_start><jupyter_text>Airport weekly penalty for arrivals and departures delays In this notebook we compute a weekly "**penalty**" score for each airport that depends on both the its incoming and outgoing flights. The score adds `0.5` for each incoming flight that is more than _15 minutes_ late, and `1` foreach outgoing flight that is more than _15 minutes_ late.Formally speaking, let $w = \{1, 2, \ldots, 52 \}$ be the _week number_ in a year $y$ and $f$ be a flight _leaving from_ or _arriving to_ $airport$ on week $w$ of year $year$. Then we compute the weekly penalties as the following: $$ ArrivalPenalty(f, y, w) = \begin{cases} 0.5 & \mbox{if } ArrDelay(f) > 15\mbox{ seconds}\\                                           0 & \mbox{otherwise}                                 \end{cases}$$ $$ DeparturePenalty(f, y, w) = \begin{cases} 1 & \mbox{if } DepDelay(f) > 15\mbox{ seconds}\\                                             0 & \mbox{otherwise}                                   \end{cases}$$ $$ WeeklyPenalty(airport, y, w) = \sum_{f} ArrivalPenalty(f, y, w) + DeparturePenalty(f, y, w) \\$$  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/04_cache_airport_weekly_penalty.parquet'
missings_dataframe_path = '../dataset/04_cache_airport_weekly_penalty_missings.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 pyspark.sql import SparkSession
import pyspark.sql.functions as F
from pyspark.sql.types import *

from pathlib import Path

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 = 'Airport weekly penalty'
spark = SparkSession  \
    .builder          \
    .appName(appName) \
    .master(master)   \
    .getOrCreate()<jupyter_output><empty_output><jupyter_text>Load dataTry 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 DEV:
    # DEV preprocessing
    perform_DEV_dataset_preprocessing(spark)
else:
    # Production preprocessing
    perform_dataset_preprocessing(spark)
# Load the parquet dataset
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># Keep only the dimensions we need
df = df.select('Year', 'Month', 'DayofMonth', 'ArrTime', 'DepTime', 'ArrDelay', 'DepDelay', 'Origin', 'Dest')
# Explore the data
#df.printSchema()<jupyter_output><empty_output><jupyter_text>Missing values exploration and exclusion<jupyter_code>df.count()
df, df_missing = get_df_missing_and_drop_all_null(df, False, missings_dataframe_path)
df.count()
df.printSchema()
df = df.select('Year', 'Month', 'DayofMonth', 'ArrDelay', 'DepDelay', 'Origin', 'Dest')
# Explore the data
df.printSchema()
# cast delays to integer
df = df.withColumn('ArrDelay', df['ArrDelay'].cast(IntegerType()))
df = df.withColumn('DepDelay', df['DepDelay'].cast(IntegerType()))
df.printSchema()<jupyter_output>root
 |-- Year: integer (nullable = true)
 |-- Month: integer (nullable = true)
 |-- DayofMonth: integer (nullable = true)
 |-- ArrDelay: integer (nullable = true)
 |-- DepDelay: integer (nullable = true)
 |-- Origin: string (nullable = true)
 |-- Dest: string (nullable = true)<jupyter_text>Compute weekly-penalty analytics<jupyter_code># compute the week number

# function to get the week number
def week_year(year, month, day):
    import datetime
    date = datetime.datetime(year, month, day) 
    return date.isocalendar()[1]
week_year = F.udf(week_year, IntegerType())

df = df.withColumn('WeekOfYear', week_year(df['Year'], df['Month'], df['DayofMonth']))
# Flights that are more than 15 minutes late
left_late = df.filter(df['DepDelay'] > 15)
arrived_late = df.filter(df['ArrDelay'] > 15)

# Number of times per week an airport had a departure or an arrival more than 15 minutes late
incoming_late = arrived_late.groupBy(['Year', 'WeekOfYear', arrived_late['Dest'].alias('Airport')]).count()
outgoing_late = left_late.groupBy(['Year', 'WeekOfYear', left_late['Origin'].alias('Airport')]).count()

#incoming_late.show(10)
#outgoing_late.show(10)
# Penalties on arrivals and departures
incoming_factor = 0.5
outgoing_factor = 1.0

incoming_penalty = incoming_late.select('Year', 'WeekOfYear', 'Airport', (incoming_late['count'] * incoming_factor).alias('Penalty'))
outgoing_penalty = outgoing_late.select('Year', 'WeekOfYear', 'Airport', (outgoing_late['count'] * outgoing_factor).alias('Penalty'))

#incoming_penalty.show(10)
#outgoing_penalty.show(10)
# Sum up the penalties
penalties = incoming_penalty.unionAll(outgoing_penalty)               \
                            .groupBy('Year', 'WeekOfYear', 'Airport') \
                            .sum('Penalty')                           \
                            .withColumnRenamed('sum(Penalty)', 'WeeklyPenalty')

# penalties.show(10)
# Store output Dataframe (or load it if already existing)
if not Path(transformed_dataset_path).is_dir():
    penalties.write.mode('overwrite').save(transformed_dataset_path, format='parquet')

penalties = spark.read.load(transformed_dataset_path)
# Output a list of tuples of schema:
# ('Year', 'WeekYear', 'Airport', 'WeeklyPenalty')
penalty_data = penalties.rdd.map(tuple).collect()
print(penalty_data[:10])

# Airports
airports = penalties.select('Airport').distinct().orderBy('Airport').rdd.map(lambda x : x[0]).collect()<jupyter_output>[(2007, 2, 'DTW', 784.5), (2007, 3, 'TUL', 209.0), (2007, 2, 'PSP', 96.0), (2007, 3, 'GRB', 67.0), (2007, 2, 'GTF', 12.5), (2007, 5, 'MLU', 27.0), (2007, 3, 'SJT', 13.5), (2007, 5, 'FAI', 16.0), (2007, 7, 'DAY', 148.0), (2007, 8, 'DAB', 30.0)]<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, interactive, fixed, interact_manual
import ipywidgets as widgets

import numpy as np
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt<jupyter_output><empty_output><jupyter_text>Weekly penalties over a year analytics<jupyter_code>def get_pd_dataframe(airport, years, df):
    rows = df.filter(F.col('Airport') == airport) \
             .filter(F.col('Year').isin(*years)) \
             .select('Year', 'WeekOfYear', 'WeeklyPenalty') \
             .orderBy('Year', 'WeekOfYear') \
             .collect()
    
    nb_years = len(years)
    nb_weeks = 52
    data = np.zeros((nb_weeks, nb_years))
    for row in rows:
        year = row[0] - years[0]
        week = row[1] - 1
        pen = row[2]

        if week > 51: continue
        data[week, year] = pen
    columns = [str(y) for y in years]
    indices = range(1, 53)
    res = pd.DataFrame(data=data, columns=columns, index=indices)
    return res

def plot_penalty_time_series(airport, years, df, ax):
    df = get_pd_dataframe(airport, years, df)
    title = '{} Airport - Weekly penalty'.format(airport)
    if df.empty:
        print('No data for airport {}'.format(airport))
    else:
        df.plot(title=title, grid=True, xticks=range(0, 53, 4), ax=ax)<jupyter_output><empty_output><jupyter_text>Average penalty in a year<jupyter_code>def get_average_df(airport, years, df):
    rows = df.filter(F.col('Airport') == airport) \
             .filter(F.col('Year').isin(*years)) \
             .groupBy('Year') \
             .avg('WeeklyPenalty') \
             .withColumnRenamed('avg(WeeklyPenalty)', 'AveragePenalty') \
             .select('Year', 'AveragePenalty') \
             .collect()
    
    nb_years = len(years)
    data = np.zeros(nb_years)
    for row in rows:
        year = row[0] - years[0]
        avg_pen = row[1]
        data[year] = avg_pen 
    res = pd.DataFrame({airport: data}, index=years)
    return res

def plot_average_penalty(airport, years, df, ax):
    df = get_average_df(airport, years, df)
    title = '{} Airport - Average Penalties'.format(airport)
    if df.empty:
        print('No data for airport {}'.format(airport))
    else:
        df.plot.bar(y=airport, title=title, rot=90, ax=ax, legend=False)
def ui_callback(airport, years, df):
    plt.figure(figsize=(15,12))
    plt.clf()
    ax = plt.subplot(211)
    plot_penalty_time_series(airport, range(years[0], years[1] + 1), df, ax)
    
    ax = plt.subplot(212)
    plot_average_penalty(airport, 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 = range(1994, 2009)
years = [(str(y), y) for y in years]
years_w = widgets.SelectionRangeSlider(options=years,
                                       index=(0, 2),
                                       description='Years',
                                       continuous_update=False)
# Airport selection menu
airports_w = widgets.Dropdown(options=airports,
                              value=airports[0],
                              description='Airport')

out = widgets.interactive_output(ui_callback, {'airport': airports_w, 'years': years_w, 'df': widgets.fixed(penalties)})
ui = widgets.HBox([airports_w, years_w])<jupyter_output><empty_output><jupyter_text>Data VisualizationAnalytics for airports weekly penalty are reported below.  Missing values and valid data<jupyter_code>plot_missing_values_stacked_bar(df_missing)<jupyter_output><empty_output><jupyter_text>Airports weekly penaltyA **line plot** is used to display penalties as a time series. On the `x` axis the week number is reported, while on the `y` axis we show the weekly penalty.Moreover, a **bar plot** is chosen to display the *yearly average* weekly-penalty of a given airport.<jupyter_code>display(ui, out)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12751232461130071, 'fraction_numerical': 0.018960940462646948, 'mean_word_length': 3.797180536607549, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 51, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 9, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '315008', 'license': 'MIT', 'n_tokens_mistral': 3612, 'n_tokens_neox': 3354, 'n_words': 888, 'path': 'notebooks/04-weekly-airport-penalty.ipynb', 'repo_name': 'polimi-mt-acg/USA-airlines-statistics-1994-2008', 'size': '10548'}