Running sparklyr – RStudio’s R Interface to Spark on Amazon EMR

# load the sparklyr, dplyr, ggplot2, and DBI packages
library(sparklyr)
library(dplyr)
library(ggplot2)
library(DBI)

# create the Spark context
sc <- spark_connect(master = "yarn", version = "3.2.1")


# copy some builtin sample data to the Spark cluster
iris_tbl <- copy_to(sc, iris)
flights_tbl <- copy_to(sc, nycflights13::flights, "flights")
batting_tbl <- copy_to(sc, Lahman::Batting, "batting")
src_tbls(sc)

# filter by departure delay and print the first few records
flights_tbl %>% filter(dep_delay == 2)

# plot data on flight delays
delay <- flights_tbl %>% 
  group_by(tailnum) %>%
  summarise(count = n(), dist = mean(distance), delay = mean(arr_delay)) %>%
  filter(count > 20, dist < 2000, !is.na(delay)) %>%
  collect


ggplot(delay, aes(dist, delay)) +
  geom_point(aes(size = count), alpha = 1/2) +
  geom_smooth() +
  scale_size_area(max_size = 2)

# window functions  
batting_tbl %>%
  select(playerID, yearID, teamID, G, AB:H) %>%
  group_by(playerID) %>%
  filter(min_rank(desc(H)) <= 2 & H > 0) %>%
  arrange(playerID, yearID, teamID)

# use SQL  
iris_preview <- dbGetQuery(sc, "SELECT * FROM iris LIMIT 10")
iris_preview

# Machine learning example using Spark MLlib

# copy the mtcars data into Spark
mtcars_tbl <- copy_to(sc, mtcars)

# transform our data set, and then partition into 'training', 'test'
partitions <- mtcars_tbl %>%
  filter(hp >= 100) %>%
  mutate(cyl8 = cyl == 8) %>%
  sdf_random_split(training = 0.5, test = 0.5, seed = 1099)

# fit a linear regression model to the training dataset
fit <- partitions$training %>%
  ml_linear_regression(response = "mpg", features = c("wt", "cyl"))
fit

summary(fit)

# get the 10th row in test data
car <- tbl_df(partitions$test) %>% slice(10)
# predict the mpg 
predicted_mpg <- car$cyl * fit$coefficients["cyl"] + car$wt * fit$coefficients["wt"] + fit$coefficients["(Intercept)"]

# print the original and the predicted
sprintf("original mpg = %s, predicted mpg = %s", car$mpg, predicted_mpg)

# app.R
library(shiny)
library(ggplot2)
library(sparklyr)
library(dplyr)
library(nycflights13)

options(bitmapType='cairo')

# create the Spark context
sc <- spark_connect(master = "yarn", version = "3.2.1")

flights_tbl <- copy_to(sc, nycflights13::flights, "flights")
delay <- flights_tbl %>%
    group_by(tailnum) %>%
    summarise(count = n(), dist = mean(distance), delay = mean(arr_delay)) %>%
    filter(count > 20, dist < 2000, !is.na(delay)) %>%
    collect

server <- function(input, output) {
    output$flightDelayPlot <- renderPlot({
        ggplot(delay, aes(dist, delay)) +
            geom_point(aes(size = count), alpha = 1/2) +
            geom_smooth() +
            scale_size_area(max_size = 2)
    })
}

# UI
ui <- fluidPage(
    titlePanel("flight delay by distance"),
    mainPanel(
        plotOutput("flightDelayPlot")
    )
)
shinyApp(ui, server)