Scaling R with Spark
using sparklyr
Javier Luraschi
01/17/2019
1 / 32
What to do when code is slow?
mtcars %>% lm(mpg ~ wt + cyl, .)2 / 32
What to do when code is slow?
mtcars %>% lm(mpg ~ wt + cyl, .)# Samplemtcars %>% dplyr::sample_n(10) %>% lm(mpg ~ wt + cyl, .)3 / 32
What to do when code is slow?
mtcars %>% lm(mpg ~ wt + cyl, .)# Samplemtcars %>% dplyr::sample_n(10) %>% lm(mpg ~ wt + cyl, .)# Profileprofvis::profvis(mtcars %>% lm(mpg ~ wt + cyl, .))4 / 32
What to do when code is slow?
mtcars %>% lm(mpg ~ wt + cyl, .)# Samplemtcars %>% dplyr::sample_n(10) %>% lm(mpg ~ wt + cyl, .)# Profileprofvis::profvis(mtcars %>% lm(mpg ~ wt + cyl, .))# Scale Upcloudml::cloudml_train("train.R")5 / 32
What to do when code is slow?
mtcars %>% lm(mpg ~ wt + cyl, .)# Samplemtcars %>% dplyr::sample_n(10) %>% lm(mpg ~ wt + cyl, .)# Profileprofvis::profvis(mtcars %>% lm(mpg ~ wt + cyl, .))# Scale Upcloudml::cloudml_train("train.R")# Scale Outmtcars_tbl %>% sparklyr::ml_linear_regression(mpg ~ wt + cyl)6 / 32
What to do when code is slow?
mtcars %>% lm(mpg ~ wt + cyl, .)# Samplemtcars %>% dplyr::sample_n(10) %>% lm(mpg ~ wt + cyl, .)# Profileprofvis::profvis(mtcars %>% lm(mpg ~ wt + cyl, .))# Scale Upcloudml::cloudml_train("train.R")# Scale Outmtcars_tbl %>% sparklyr::ml_linear_regression(mpg ~ wt + cyl)Note: There are many more ways to sample, scale-up and scale-out.
7 / 32
Scaling Out with R and Spark
# Scale Outmtcars_tbl %>% sparklyr::ml_linear_regression(mpg ~ wt + cyl)8 / 32
How to use R with Spark?
library(sparklyr) # R Interface to Apache Sparkspark_install() # Install Apache Sparksc <- spark_connect(master = "local") # Connect to Spark cluster9 / 32
How to use R with Spark?
library(sparklyr) # R Interface to Apache Sparkspark_install() # Install Apache Sparksc <- spark_connect(master = "local") # Connect to Spark clustercars_tbl <- spark_read_csv(sc, "cars", "mtcars/") # Read data in Sparksummarize(cars_tbl, n = n()) # Count records with dplyrdbGetQuery(sc, "SELECT count(*) FROM cars") # Count records with DBI10 / 32
How to use R with Spark?
library(sparklyr) # R Interface to Apache Sparkspark_install() # Install Apache Sparksc <- spark_connect(master = "local") # Connect to Spark clustercars_tbl <- spark_read_csv(sc, "cars", "mtcars/") # Read data in Sparksummarize(cars_tbl, n = n()) # Count records with dplyrdbGetQuery(sc, "SELECT count(*) FROM cars") # Count records with DBIml_linear_regression(cars_tbl, mpg ~ wt + cyl) # Perform linear regression11 / 32
How to use R with Spark?
library(sparklyr) # R Interface to Apache Sparkspark_install() # Install Apache Sparksc <- spark_connect(master = "local") # Connect to Spark clustercars_tbl <- spark_read_csv(sc, "cars", "mtcars/") # Read data in Sparksummarize(cars_tbl, n = n()) # Count records with dplyrdbGetQuery(sc, "SELECT count(*) FROM cars") # Count records with DBIml_linear_regression(cars_tbl, mpg ~ wt + cyl) # Perform linear regressionpipeline <- ml_pipeline(sc) %>% # Define Spark pipeline ft_r_formula(mpg ~ wt + cyl) %>% # Add formula transformation ml_linear_regression() # Add model to pipelinefitted <- ml_fit(pipeline, cars_tbl) # Fit pipeline12 / 32
How to use R with Spark?
library(sparklyr) # R Interface to Apache Sparkspark_install() # Install Apache Sparksc <- spark_connect(master = "local") # Connect to Spark clustercars_tbl <- spark_read_csv(sc, "cars", "mtcars/") # Read data in Sparksummarize(cars_tbl, n = n()) # Count records with dplyrdbGetQuery(sc, "SELECT count(*) FROM cars") # Count records with DBIml_linear_regression(cars_tbl, mpg ~ wt + cyl) # Perform linear regressionpipeline <- ml_pipeline(sc) %>% # Define Spark pipeline ft_r_formula(mpg ~ wt + cyl) %>% # Add formula transformation ml_linear_regression() # Add model to pipelinefitted <- ml_fit(pipeline, cars_tbl) # Fit pipelinespark_context(sc) %>% invoke("version") # Extend sparklyr with Scalaspark_apply(cars_tbl, nrow) # Extend sparklyr with R13 / 32
What about realtime data?
14 / 32
Spark Structured Streams
cars_str <- stream_read_csv(sc, "mtcars/", "cars") # Read stream in Spark15 / 32
Spark Structured Streams
cars_str <- stream_read_csv(sc, "mtcars/", "cars") # Read stream in Sparkout_str <- summarize(cars_str, n = n()) # Count records with dplyrout_str <- dbGetQuery(sc, "SELECT count(*) FROM cars") # Count records with DBI16 / 32
Spark Structured Streams
cars_str <- stream_read_csv(sc, "mtcars/", "cars") # Read stream in Sparkout_str <- summarize(cars_str, n = n()) # Count records with dplyrout_str <- dbGetQuery(sc, "SELECT count(*) FROM cars") # Count records with DBIout_str <- ml_transform(fitted, cars_str) # Apply pipeline to stream17 / 32
Spark Structured Streams
cars_str <- stream_read_csv(sc, "mtcars/", "cars") # Read stream in Sparkout_str <- summarize(cars_str, n = n()) # Count records with dplyrout_str <- dbGetQuery(sc, "SELECT count(*) FROM cars") # Count records with DBIout_str <- ml_transform(fitted, cars_str) # Apply pipeline to streamout_str <- spark_apply(cars_str, nrow) # Extend streams with R18 / 32
Spark Structured Streams
cars_str <- stream_read_csv(sc, "mtcars/", "cars") # Read stream in Sparkout_str <- summarize(cars_str, n = n()) # Count records with dplyrout_str <- dbGetQuery(sc, "SELECT count(*) FROM cars") # Count records with DBIout_str <- ml_transform(fitted, cars_str) # Apply pipeline to streamout_str <- spark_apply(cars_str, nrow) # Extend streams with Rstream_write_csv(out_str, "output/") # Write as a CSV streamreactiveSpark(out_str) # Use as a Shiny reactive19 / 32
Streaming with Spark, Kafka and Shiny
Apache Kafka is an open-source stream-processing software platform that provides a unified, high-throughput and low-latency for handling real-time data feeds.
20 / 32
Streaming with Spark, Kafka and Shiny
Apache Kafka is an open-source stream-processing software platform that provides a unified, high-throughput and low-latency for handling real-time data feeds.
21 / 32
What's new in Spark and R?
Streams, MLeap, Kubernetes and RStudio 1.2 integration.
22 / 32
What's new in Spark and R?
Streams, MLeap, Kubernetes and RStudio 1.2 integration.
23 / 32
What are we currently working on?
- Faster data transfer with Apache Arrow: pull/1611.
- XGBoost on Spark: rstudio/sparkxgb.
25 / 32
Arrow on Spark
Arrow is a cross-language development platform for in-memory data.
26 / 32
Arrow on Spark
Arrow is a cross-language development platform for in-memory data.
devtools::install_github("apache/arrow", subdir = "r", ref = "dc5df8f")devtools::install_github("rstudio/sparklyr")library(arrow)library(sparklyr)27 / 32
Arrow on Spark
Arrow is a cross-language development platform for in-memory data.
devtools::install_github("apache/arrow", subdir = "r", ref = "dc5df8f")devtools::install_github("rstudio/sparklyr")library(arrow)library(sparklyr)
28 / 32
XGBoost on Spark
sparkxgb is a sparklyr extension that provides an interface to XGBoost on Spark.
29 / 32
XGBoost on Spark
sparkxgb is a sparklyr extension that provides an interface to XGBoost on Spark.
devtools::install_github("rstudio/sparkxgb")library(sparkxgb)30 / 32
XGBoost on Spark
sparkxgb is a sparklyr extension that provides an interface to XGBoost on Spark.
devtools::install_github("rstudio/sparkxgb")library(sparkxgb)iris_tbl <- sdf_copy_to(sc, iris)xgb_model <- xgboost_classifier( iris_tbl, Species ~ ., objective = "multi:softprob", num_class = 3, num_round = 50, max_depth = 4)xgb_model %>% ml_predict(iris_tbl) %>% glimpse()31 / 32
Thank you!
Docs: spark.rstudio.com
Blog: blog.rstudio.com/tags/sparklyr
R Help: community.rstudio.com
Spark Help: stackoverflow.com/tags/sparklyr
Issues: github.com/rstudio/sparklyr/issues
Chat: gitter.im/rstudio.sparklyr
32 / 32