# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# ## **Airport daily percentage of cancelled flights**
#

# In this notebook we compute the percentage of cancelled flights per day, throughout the entire data set.
#
# \begin{equation}
# PercentageOfCancelledFlights(dayOfYear) = \frac{CancelledFlights(dayOfYear)}{NumberOfFlights(dayOfYear)} * 100
# \end{equation}

# #### Execution mode
#
# If 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.

# +
# 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/01_cache_daily_cancelled_analytics.parquet'
missings_dataframe_path = '../dataset/01_cache_daily_cancelled_analytics_missings.pkl'
# -

# #### Find and import Apache Spark on current machine

import findspark
findspark.init()

# +
# imports
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 *
# -

# #### Create a new session

# Build a Spark SQL Session for DataFrames
master = 'local[{}]'.format(spark_local_threads)
appName = 'Daily cancelled filghts'
spark = SparkSession  \
    .builder          \
    .appName(appName) \
    .master(master)   \
    .getOrCreate()

# ### Load data
#
# If there exists a cached version of the already transformed dataset, just load it and skip all the unnecessary computation.

# +
do_computation = True

if not DEV and Path(transformed_dataset_path).is_dir():
    df = spark.read.load(transformed_dataset_path)
    do_computation = False
# -

# 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.

# +
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)

# ### Attributes selection

if do_computation:
    # Keep only the dimensions we need
    df = df.select(df['Year'], df['Month'], df['DayofMonth'], df['Cancelled'])
    # Explore the data
    df.printSchema()

# +
# get years from dataset
if do_computation:
    df_years = df.select('Year').distinct().toPandas()['Year']
else:
    df_years = list(df.columns)
    df_years.remove('DayOfMonth')
    df_years.remove('Month')

df_years = sorted(df_years)
print(df_years)
# -

# ### Missing values exploration and exclusion

df.count()

df, df_missing = get_df_missing_and_drop_all_null(df, do_computation, missings_dataframe_path)

df.count()

# ### Compute daily-percentage analytics

if do_computation:
    df.describe('Cancelled').show()
    df.describe('Year').show()

if do_computation:
    df = df.withColumn("Count", F.lit(1))

    df = df.groupBy(['Year', 'Month', 'DayOfMonth'])                \
           .sum('Count', 'Cancelled')                               \
           .withColumnRenamed('sum(Count)', 'Count')                \
           .withColumnRenamed('sum(Cancelled)', 'Cancelled')

    df.show(10)

# #### Leap years management

leap_years = range(1996,2009,4)
list(leap_years)

# insert a zero flights count and a zero cancelled flights count for all the leap years
if do_computation:
    for i in range(1994, 2009):
        if i not in leap_years:
            new_row = spark.createDataFrame([[i, 2, 29, 0, 0]])
            df = df.union(new_row)

# Compute the percentage
if do_computation:
    df = df.withColumn('DailyCancelledFlightsPercentage', (df.Cancelled / df.Count) * 100)
    df = df.select('Year', 'Month', 'DayOfMonth', 'DailyCancelledFlightsPercentage')

if do_computation:
    df.show(20)

# #### Data format optimazion

# dataset transformation so to have:
# - for each year a column with the daily cancelled percentages 
# - have the dataset rows beign 366: days in a year
if do_computation:
    for y in df_years:
        df = df.withColumn(str(y) + '_val', df.Year == y)
        df = df.withColumn(str(y) + '_val', df[str(y) + '_val'].cast(IntegerType()))
        df = df.withColumn(str(y), df[str(y) + '_val'] * df.DailyCancelledFlightsPercentage)
        df = df.drop(str(y) + '_val')

    df = df.drop('DailyCancelledFlightsPercentage')

    # do the sum so that the columns will shrink all to the same size of 366 rows
    df = df.groupBy('Month', 'DayOfMonth')                     \
           .sum()                                              \
           .drop('sum(Year)', 'sum(Month)', 'sum(DayOfMonth)') \
           .orderBy('Month', 'DayOfMonth')

    # rename colums as year name
    for y in df_years:
        df = df.withColumnRenamed('sum({})'.format(y), str(y))
    

if do_computation:
    df.write.mode('overwrite').save(transformed_dataset_path, format='parquet')

df.printSchema()

df.select('1994', '1996', '2001', '2007').show(10)

pdf = df.toPandas()

# +
# # rows?
assert(366 == pdf.shape[0])
pdf.shape
# -

# update index to match the day of the year!
pdf.index = pdf.index + 1

#pdf.head(65)
pdf.head()

pdf.tail()

# ##### Data visualization functions
#
# Scroll down to see the charts

# +
# 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
import pandas as pd
import numpy as np

import calendar
# -

fig_size=[15,10]
title_size=20


# +
def get_years_df(years):
    """
    Return the dataframe of the selected years 
    """
    years = [str(y) for y in range(*years)]
    return pdf[years]


def plot_cancelled_time_series(years):
    # select requested years from dataframe 
    df = get_years_df(years)
    
    # xticks are the last day of the month counted as day of the year
    xticks = [0]
    for m in range(1,13):
        new_val = xticks[-1] + calendar.monthrange(2016, m)[1]
        xticks.append(new_val)

    # xlabels are the months abbreviations
    xlabels = list(calendar.month_abbr)[1:]

    plt.rcParams['figure.figsize'] = fig_size
    plt.rcParams['axes.titlesize'] = title_size
    fig, ax = plt.subplots()
    
    ax.set_xticks(xticks)
    ax.set_xticklabels(xlabels) 
    
    ax.set_xlabel('Months', size=16)
    ax.set_ylabel('Cancellation percentage', size=16)
    ax.set_title('Daily cancelled flights percentage', pad=20)

    plt.xlim(xmin=-4, xmax=370)
    df.plot(grid=True, rot=45, ax=ax)
    plt.legend(loc='best')
    

def plot_average_cancelled_flights(years):
    # select requested years from dataframe 
    df = get_years_df(years)    
    averages = []
    for y in range(*years):
        averages.append(df[str(y)].mean())

    labels = [str(y) for y in range(*years)]
    
    plt.rcParams['figure.figsize'] = fig_size
    plt.rcParams['axes.titlesize'] = title_size
    fig, ax = plt.subplots()
    indexes = np.arange(len(averages))
    
    rects = ax.bar(indexes, averages, label=labels, color='orangered')

    # add percentages on bars
    for rect in rects:
        height = rect.get_height()
        ax.text(rect.get_x() + rect.get_width()/2., height - 0.18,
                str(np.round(height, decimals=2)) + '%',
                ha='center', va='bottom', color='w', size=13)
        
    ax.set_xlabel('Years', size=16)
    ax.set_ylabel('Cancellation percentage', size=16)
    ax.set_title('Average of cancelled flights percentages', pad=20)
    
    ax.set_xticks(indexes)
    ax.set_xticklabels(labels)
    ax.xaxis.set_tick_params(labelsize=14)
    ax.legend(loc='best', labels=['Cancellation percentage'])
    plt.show()


# +
def ui_callback(years):
    years = (int(years[0]), int(years[1]) + 1)  #due to range() function
    
    plot_cancelled_time_series(years)

    plot_average_cancelled_flights(years)


# Years selection range
years_w = widgets.SelectionRangeSlider(options=[(str(y), y) for y in df_years],
                                       index=(0, 2),
                                       description='Years',
                                       continuous_update=False)
ui = widgets.HBox([years_w])
# -

# ## Data Visualization
#
# Analytics showhing the percentage of daily cancelled flights are reported below.

# ### Missing values and valid data

plot_missing_values_stacked_bar(df_missing)

# ### Percentage of daily cancelled flights
#
# A line plot was used to display percentages as a time series. 
#
# - _x axis_ is the day number, from day 1 (1st of January of the selected year) to 366 (30th of December of the selected year)
# - _y axis_ reports the daily percentage of cancelled flights
#
# Moreover, also bar plot is generated to display the yearly average of cancelled flights.
#
#
# #### Data discoveries
#
# - The analytics is performed over the totality of the data. There're no missing values for the analized attributes.

out = widgets.interactive_output(ui_callback, {'years': years_w})
display(ui, out)

StarcoderdataJupyterScriptsDedupFiltered