# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .r
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: R
#     language: R
#     name: ir
# ---

Sys.setenv(SPARK_HOME="/Users/jonathandinu/spark-1.4.1-bin-hadoop2.4/")

.libPaths(c(file.path(Sys.getenv("SPARK_HOME"), "R", "lib"), .libPaths())) 

library(SparkR)

vec <- c(1,2,3)

typeof(vec)

sc = sparkR.init(sparkPackages='com.databricks:spark-csv_2.11:1.0.3')

sqlContext <- sparkRSQL.init(sc)

# Replace <AWS_ACCESS_KEY_ID> and <AWS_SECRET_ACCESS_KEY> with your credentials
link <- 's3n://<AWS_ACCESS_KEY_ID>:<AWS_SECRET_ACCESS_KEY>@mortar-example-data/airline-data'

# [Instructions to get AWS credentials](http://www.slideshare.net/jonathandinu/the-data-scientists-guide-to-apache-spark/113?src=clipshare)
#
# You can also download a ZIP of the airline data here: http://hopelessoptimism.com/static/data/airline-data
#
# Or find the original (unparsed) files here: http://www.transtats.bts.gov/DL_SelectFields.asp?Table_ID=236

flights <- read.df(sqlContext, link, source = "com.databricks.spark.csv", header='true')

head(flights)

printSchema(flights)

subset <- select(flights, flights$DEP_DELAY, flights$ARR_DELAY, 
                 flights$ORIGIN_AIRPORT_ID, flights$DEST_AIRPORT_ID)

persist(subset, 'MEMORY_AND_DISK')

head(subset)

dtypes(subset)

help(cast)

subset$DEP_DELAY <- cast(flights$DEP_DELAY, 'double')

subset$ARR_DELAY <- cast(flights$ARR_DELAY, 'double')

head(subset)

dtypes(subset)

groupedDest <- groupBy(subset, subset$DEST_AIRPORT_ID)

groupedOrigin <- groupBy(subset, subset$ORIGIN_AIRPORT_ID)

dest_mean <- mean(groupedDest) 
origin_mean <- mean(groupedOrigin) 

persist(dest_mean, "MEMORY_AND_DISK")
persist(origin_mean, "MEMORY_AND_DISK")

head(dest_mean)

head(arrange(dest_mean, dest_mean$'AVG(ARR_DELAY)'))

head(arrange(dest_mean, desc(dest_mean$'AVG(ARR_DELAY)')))

dest_mean <- withColumnRenamed(dest_mean, 'AVG(ARR_DELAY)', 'avg_arr_delay')

head(dest_mean)

arrange(dest_mean, desc(dest_mean$avg_arr_delay))

# ## Big Data... Small Learning

# install.packages("ggplot2")
library(ggplot2)

local_df <- collect(dest_mean)

#ggplot(data=local_df, aes(x='Delay (min)')) + geom_histogram(aes(fill=..count..))
ggplot(data=local_df, aes(x=local_df$avg_arr_delay)) + geom_histogram(aes(fill=..count..)) + ggtitle("Average Flight Delay") + xlab('Delay (min)')

# ## Putting it All together

# +
# use Spark CSV to load data
flights <- read.df(sqlContext, link, source = "com.databricks.spark.csv", header='true')

# subset columns needed
subset <- select(flights, flights$DEP_DELAY, flights$ARR_DELAY, 
                 flights$ORIGIN_AIRPORT_ID, flights$DEST_AIRPORT_ID)

# cast delay columns to double type
subset$DEP_DELAY <- cast(flights$DEP_DELAY, 'double')
subset$ARR_DELAY <- cast(flights$ARR_DELAY, 'double')

# group by airport and compute mean
groupedDest <- groupBy(subset, subset$DEST_AIRPORT_ID)
groupedOrigin <- groupBy(subset, subset$ORIGIN_AIRPORT_ID)

dest_mean <- mean(groupedDest) 
origin_mean <- mean(groupedOrigin) 

# convert SparkR dataframe to R dataframe
dest_mean <- withColumnRenamed(dest_mean, 'AVG(ARR_DELAY)', 'avg_arr_delay')
local_df <- collect(dest_mean)

# visualize with ggplot2
ggplot(data=local_df, aes(x=local_df$avg_arr_delay)) + geom_histogram(aes(fill=..count..)) + ggtitle("Average Flight Delay") + xlab('Delay (min)')

StarcoderdataJupyterScriptsDedupFiltered