Apache Spark is a fast and general-purpose cluster computing system.
It provides high-level APIs in Java, Scala, Python and R, and an optimized engine that supports general execution graphs.
It also supports a rich set of higher-level tools including Spark SQL for SQL and structured data processing, MLlib for machine learning, GraphX for graph processing, and Spark Streaming.
http://spark.apache.org/docs/latest/index.html
Connect to Spark from R. The sparklyr package provides a
complete dplyr backend.
Filter and aggregate Spark datasets then bring them into R for
analysis and visualization.
Use Spark's distributed machine learning library from R.
Create extensions that call the full Spark API and provide
interfaces to Spark packages.
Installation
You can install the sparklyr package from CRAN as follows:
install.packages("sparklyr")
You should also install a local version of Spark for development purposes:
library(sparklyr)
spark_install(version = "2.1.0")
To upgrade to the latest version of sparklyr, run the following command and restart your r session:
devtools::install_github("rstudio/sparklyr")
If you use the RStudio IDE, you should also download the latest preview release of the IDE which includes several enhancements for interacting with Spark (see the RStudio IDE section below for more details).
Connecting to Spark
You can connect to both local instances of Spark as well as remote Spark clusters. Here we'll connect to a local instance of Spark via the spark_connect function:
library(sparklyr)
sc <- spark_connect(master = "local")
The returned Spark connection (sc) provides a remote dplyr data source to the Spark cluster.
For more information on connecting to remote Spark clusters see the Deployment section of the sparklyr website.
Using dplyr
We can now use all of the available dplyr verbs against the tables within the cluster.
We'll start by copying some datasets from R into the Spark cluster (note that you may need to install the nycflights13 and Lahman packages in order to execute this code):
install.packages(c("nycflights13", "Lahman"))library(dplyr)
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)## [1] "batting" "flights" "iris"
To start with here's a simple filtering example:
# filter by departure delay and print the first few records
flights_tbl %>% filter(dep_delay == 2)## # Source: lazy query [?? x 19]
## # Database: spark_connection
## year month day dep_time sched_dep_time dep_delay arr_time
## <int> <int> <int> <int> <int> <dbl> <int>
## 1 2013 1 1 517 515 2 830
## 2 2013 1 1 542 540 2 923
## 3 2013 1 1 702 700 2 1058
## 4 2013 1 1 715 713 2 911
## 5 2013 1 1 752 750 2 1025
## 6 2013 1 1 917 915 2 1206
## 7 2013 1 1 932 930 2 1219
## 8 2013 1 1 1028 1026 2 1350
## 9 2013 1 1 1042 1040 2 1325
## 10 2013 1 1 1231 1229 2 1523
## # ... with more rows, and 12 more variables: sched_arr_time <int>,
## # arr_delay <dbl>, carrier <chr>, flight <int>, tailnum <chr>,
## # origin <chr>, dest <chr>, air_time <dbl>, distance <dbl>, hour <dbl>,
## # minute <dbl>, time_hour <dbl>Introduction to dplyr provides additional dplyr examples you can try. For example, consider the last example from the tutorial which plots 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
# plot delays
library(ggplot2)
ggplot(delay, aes(dist, delay)) +
geom_point(aes(size = count), alpha = 1/2) +
geom_smooth() +
scale_size_area(max_size = 2)## `geom_smooth()` using method = 'gam'
Window Functions
dplyr window functions are also supported, for example:
batting_tbl %>%
select(playerID, yearID, teamID, G, AB:H) %>%
arrange(playerID, yearID, teamID) %>%
group_by(playerID) %>%
filter(min_rank(desc(H)) <= 2 & H > 0)## # Source: lazy query [?? x 7]
## # Database: spark_connection
## # Groups: playerID
## # Ordered by: playerID, yearID, teamID
## playerID yearID teamID G AB R H
## <chr> <int> <chr> <int> <int> <int> <int>
## 1 aaronha01 1959 ML1 154 629 116 223
## 2 aaronha01 1963 ML1 161 631 121 201
## 3 abbotji01 1999 MIL 20 21 0 2
## 4 abnersh01 1992 CHA 97 208 21 58
## 5 abnersh01 1990 SDN 91 184 17 45
## 6 acklefr01 1963 CHA 2 5 0 1
## 7 acklefr01 1964 CHA 3 1 0 1
## 8 adamecr01 2016 COL 121 225 25 49
## 9 adamecr01 2015 COL 26 53 4 13
## 10 adamsac01 1943 NY1 70 32 3 4
## # ... with more rows
For additional documentation on using dplyr with Spark see the dplyr section of the sparklyr website.
Using SQL
It's also possible to execute SQL queries directly against tables within a Spark cluster. The spark_connection object implements a DBI interface for Spark, so you can use dbGetQuery to execute SQL and return the result as an R data frame:
library(DBI)
iris_preview <- dbGetQuery(sc, "SELECT * FROM iris LIMIT 10")
iris_preview## Sepal_Length Sepal_Width Petal_Length Petal_Width Species
## 1 5.1 3.5 1.4 0.2 setosa
## 2 4.9 3.0 1.4 0.2 setosa
## 3 4.7 3.2 1.3 0.2 setosa
## 4 4.6 3.1 1.5 0.2 setosa
## 5 5.0 3.6 1.4 0.2 setosa
## 6 5.4 3.9 1.7 0.4 setosa
## 7 4.6 3.4 1.4 0.3 setosa
## 8 5.0 3.4 1.5 0.2 setosa
## 9 4.4 2.9 1.4 0.2 setosa
## 10 4.9 3.1 1.5 0.1 setosa
Machine Learning
You can orchestrate machine learning algorithms in a Spark cluster via the machine learning functions within sparklyr. These functions connect to a set of high-level APIs built on top of DataFrames that help you create and tune machine learning workflows.
Here's an example where we use ml_linear_regression to fit a linear regression model. We'll use the built-in mtcars dataset, and see if we can predict a car's fuel consumption (mpg) based on its weight (wt), and the number of cylinders the engine contains (cyl). We'll assume in each case that the relationship between mpg and each of our features is linear.
# copy mtcars 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_partition(training = 0.5, test = 0.5, seed = 1099)
# fit a linear model to the training dataset
fit <- partitions$training %>%
ml_linear_regression(response = "mpg", features = c("wt", "cyl"))
fit## Call: ml_linear_regression.tbl_spark(., response = "mpg", features = c("wt", "cyl"))
##
## Formula: mpg ~ wt + cyl
##
## Coefficients:
## (Intercept) wt cyl
## 33.499452 -2.818463 -0.923187
For linear regression models produced by Spark, we can use summary() to learn a bit more about the quality of our fit, and the statistical significance of each of our predictors.
summary(fit)## Call: ml_linear_regression.tbl_spark(., response = "mpg", features = c("wt", "cyl"))
##
## Deviance Residuals:
## Min 1Q Median 3Q Max
## -1.752 -1.134 -0.499 1.296 2.282
##
## Coefficients:
## (Intercept) wt cyl
## 33.499452 -2.818463 -0.923187
##
## R-Squared: 0.8274
## Root Mean Squared Error: 1.422
Spark machine learning supports a wide array of algorithms and feature transformations and as illustrated above it's easy to chain these functions together with dplyr pipelines. To learn more see the machine learning section.
Reading and Writing Data
You can read and write data in CSV, JSON, and Parquet formats. Data can be stored in HDFS, S3, or on the local filesystem of cluster nodes.
temp_csv <- tempfile(fileext = ".csv")
temp_parquet <- tempfile(fileext = ".parquet")
temp_json <- tempfile(fileext = ".json")
spark_write_csv(iris_tbl, temp_csv)
iris_csv_tbl <- spark_read_csv(sc, "iris_csv", temp_csv)
spark_write_parquet(iris_tbl, temp_parquet)
iris_parquet_tbl <- spark_read_parquet(sc, "iris_parquet", temp_parquet)
spark_write_json(iris_tbl, temp_json)
iris_json_tbl <- spark_read_json(sc, "iris_json", temp_json)
src_tbls(sc)## [1] "batting" "flights" "iris" "iris_csv"
## [5] "iris_json" "iris_parquet" "mtcars"
Distributed R
You can execute arbitrary r code across your cluster using spark_apply. For example, we can apply rgamma over iris as follows:
spark_apply(iris_tbl, function(data) {
data[1:4] + rgamma(1,2)
})## # Source: table<sparklyr_tmp_115c74acb6510> [?? x 4]
## # Database: spark_connection
## Sepal_Length Sepal_Width Petal_Length Petal_Width
## <dbl> <dbl> <dbl> <dbl>
## 1 5.336757 3.736757 1.636757 0.4367573
## 2 5.136757 3.236757 1.636757 0.4367573
## 3 4.936757 3.436757 1.536757 0.4367573
## 4 4.836757 3.336757 1.736757 0.4367573
## 5 5.236757 3.836757 1.636757 0.4367573
## 6 5.636757 4.136757 1.936757 0.6367573
## 7 4.836757 3.636757 1.636757 0.5367573
## 8 5.236757 3.636757 1.736757 0.4367573
## 9 4.636757 3.136757 1.636757 0.4367573
## 10 5.136757 3.336757 1.736757 0.3367573
## # ... with more rows
You can also group by columns to perform an operation over each group of rows and make use of any package within the closure:
spark_apply(
iris_tbl,
function(e) broom::tidy(lm(Petal_Width ~ Petal_Length, e)),
names = c("term", "estimate", "std.error", "statistic", "p.value"),
group_by = "Species"
)## # Source: table<sparklyr_tmp_115c73965f30> [?? x 6]
## # Database: spark_connection
## Species term estimate std.error statistic p.value
## <chr> <chr> <dbl> <dbl> <dbl> <dbl>
## 1 versicolor (Intercept) -0.08428835 0.16070140 -0.5245029 6.023428e-01
## 2 versicolor Petal_Length 0.33105360 0.03750041 8.8279995 1.271916e-11
## 3 virginica (Intercept) 1.13603130 0.37936622 2.9945505 4.336312e-03
## 4 virginica Petal_Length 0.16029696 0.06800119 2.3572668 2.253577e-02
## 5 setosa (Intercept) -0.04822033 0.12164115 -0.3964146 6.935561e-01
## 6 setosa Petal_Length 0.20124509 0.08263253 2.4354220 1.863892e-02
Extensions
The facilities used internally by sparklyr for its dplyr and machine learning interfaces are available to extension packages. Since Spark is a general purpose cluster computing system there are many potential applications for extensions (e.g. interfaces to custom machine learning pipelines, interfaces to 3rd party Spark packages, etc.).
Here's a simple example that wraps a Spark text file line counting function with an R function:
# write a CSV
tempfile <- tempfile(fileext = ".csv")
write.csv(nycflights13::flights, tempfile, row.names = FALSE, na = ")
# define an R interface to Spark line counting
count_lines <- function(sc, path) {
spark_context(sc) %>%
invoke("textFile", path, 1L) %>%
invoke("count")
}
# call spark to count the lines of the CSV
count_lines(sc, tempfile)## [1] 336777
To learn more about creating extensions see the Extensions section of the sparklyr website.
Table Utilities
You can cache a table into memory with:
tbl_cache(sc, "batting")
and unload from memory using:
tbl_uncache(sc, "batting")
Connection Utilities
You can view the Spark web console using the spark_web function:
spark_web(sc)
You can show the log using the spark_log function:
spark_log(sc, n = 10)## 17/11/09 15:55:18 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 69 (/var/folders/fz/v6wfsg2x1fb1rw4f6r0x4jwm0000gn/T//RtmpyR8oP9/file115c74b94924.csv MapPartitionsRDD[258] at textFile at NativeMethodAccessorImpl.java:0) (first 15 tasks are for partitions Vector(0))
## 17/11/09 15:55:18 INFO TaskSchedulerImpl: Adding task set 69.0 with 1 tasks
## 17/11/09 15:55:18 INFO TaskSetManager: Starting task 0.0 in stage 69.0 (TID 140, localhost, executor driver, partition 0, PROCESS_LOCAL, 4904 bytes)
## 17/11/09 15:55:18 INFO Executor: Running task 0.0 in stage 69.0 (TID 140)
## 17/11/09 15:55:18 INFO HadoopRDD: Input split: file:/var/folders/fz/v6wfsg2x1fb1rw4f6r0x4jwm0000gn/T/RtmpyR8oP9/file115c74b94924.csv:0+33313106
## 17/11/09 15:55:18 INFO Executor: Finished task 0.0 in stage 69.0 (TID 140). 832 bytes result sent to driver
## 17/11/09 15:55:18 INFO TaskSetManager: Finished task 0.0 in stage 69.0 (TID 140) in 126 ms on localhost (executor driver) (1/1)
## 17/11/09 15:55:18 INFO TaskSchedulerImpl: Removed TaskSet 69.0, whose tasks have all completed, from pool
## 17/11/09 15:55:18 INFO DAGScheduler: ResultStage 69 (count at NativeMethodAccessorImpl.java:0) finished in 0.126 s
## 17/11/09 15:55:18 INFO DAGScheduler: Job 47 finished: count at NativeMethodAccessorImpl.java:0, took 0.131380 s
Finally, we disconnect from Spark:
spark_disconnect(sc)
RStudio IDE
The latest RStudio Preview Release of the RStudio IDE includes integrated support for Spark and the sparklyr package, including tools for:
Creating and managing Spark connections
Browsing the tables and columns of Spark DataFrames
Previewing the first 1,000 rows of Spark DataFrames
Once you've installed the sparklyr package, you should find a new Spark pane within the IDE. This pane includes a New Connection dialog which can be used to make connections to local or remote Spark instances:
Once you've connected to Spark you'll be able to browse the tables contained within the Spark cluster and preview Spark DataFrames using the standard RStudio data viewer:
You can also connect to Spark through Livy through a new connection dialog:
The RStudio IDE features for sparklyr are available now as part of the RStudio Preview Release.
Using H2O
rsparkling is a CRAN package from H2O that extends sparklyr to provide an interface into Sparkling Water. For instance, the following example installs, configures and runs h2o.glm:
options(rsparkling.sparklingwater.version = "2.1.14")
library(rsparkling)
library(sparklyr)
library(dplyr)
library(h2o)
sc <- spark_connect(master = "local", version = "2.1.0")
mtcars_tbl <- copy_to(sc, mtcars, "mtcars")
mtcars_h2o <- as_h2o_frame(sc, mtcars_tbl, strict_version_check = FALSE)
mtcars_glm <- h2o.glm(x = c("wt", "cyl"),
y = "mpg",
training_frame = mtcars_h2o,
lambda_search = TRUE)mtcars_glm## Model Details:
## ==============
##
## H2ORegressionModel: glm
## Model ID: GLM_model_R_1510271749678_1
## GLM Model: summary
## family link regularization
## 1 gaussian identity Elastic Net (alpha = 0.5, lambda = 0.1013 )
## lambda_search
## 1 nlambda = 100, lambda.max = 10.132, lambda.min = 0.1013, lambda.1se = -1.0
## number_of_predictors_total number_of_active_predictors
## 1 2 2
## number_of_iterations training_frame
## 1 100 frame_rdd_29_b907d4915799eac74fb1ea60ad594bbf
##
## Coefficients: glm coefficients
## names coefficients standardized_coefficients
## 1 Intercept 38.941654 20.090625
## 2 cyl -1.468783 -2.623132
## 3 wt -3.034558 -2.969186
##
## H2ORegressionMetrics: glm
## ** Reported on training data. **
##
## MSE: 6.017684
## RMSE: 2.453097
## MAE: 1.940985
## RMSLE: 0.1114801
## Mean Residual Deviance : 6.017684
## R^2 : 0.8289895
## Null Deviance :1126.047
## Null D.o.F. :31
## Residual Deviance :192.5659
## Residual D.o.F. :29
## AIC :156.2425spark_disconnect(sc)
Connecting through Livy
Livy enables remote connections to Apache Spark clusters. Before connecting to Livy, you will need the connection information to an existing service running Livy. Otherwise, to test livy in your local environment, you can install it and run it locally as follows:
livy_install()livy_service_start()
To connect, use the Livy service address as master and method = "livy" in spark_connect. Once connection completes, use sparklyr as usual, for instance:
sc <- spark_connect(master = "http://localhost:8998", method = "livy")
copy_to(sc, iris)## # Source: table<iris> [?? x 5]
## # Database: spark_connection
## Sepal_Length Sepal_Width Petal_Length Petal_Width Species
## <dbl> <dbl> <dbl> <dbl> <chr>
## 1 5.1 3.5 1.4 0.2 setosa
## 2 4.9 3.0 1.4 0.2 setosa
## 3 4.7 3.2 1.3 0.2 setosa
## 4 4.6 3.1 1.5 0.2 setosa
## 5 5.0 3.6 1.4 0.2 setosa
## 6 5.4 3.9 1.7 0.4 setosa
## 7 4.6 3.4 1.4 0.3 setosa
## 8 5.0 3.4 1.5 0.2 setosa
## 9 4.4 2.9 1.4 0.2 setosa
## 10 4.9 3.1 1.5 0.1 setosa
## # ... with more rowsspark_disconnect(sc)
Once you are done using livy locally, you should stop this service with:
livy_service_stop()
To connect to remote livy clusters that support basic authentication connect as:
config <- livy_config(username="<username>", password="<password">)
sc <- spark_connect(master = "<address>", method = "livy", config = config)
spark_disconnect(sc)sparklyr 0.9 is now available on CRAN
SparkR is an R package that provides a light-weight frontend to use Apache Spark from R.
In Spark 2.4.5, SparkR provides a distributed data frame implementation that supports operations like selection, filtering, aggregation etc.
(similar to R data frames, dplyr) but on large datasets.
SparkR also supports distributed machine learning using MLlib.
SparkDataFrame
A SparkDataFrame is a distributed collection of data organized into named columns.
It is conceptually equivalent to a table in a relational database or a data frame in R, but with richer optimizations under the hood.
SparkDataFrames can be constructed from a wide array of sources such as:
structured data files, tables in Hive, external databases, or existing local R data frames.
All of the examples on this page use sample data included in R or the Spark distribution and can be run using the ./bin/sparkR shell.
Starting Up: SparkSession
The entry point into SparkR is the SparkSession which connects your R program to a Spark cluster.
You can create a SparkSession using sparkR.session and pass in options such as the application name, any spark packages depended on, etc.
Further, you can also work with SparkDataFrames via SparkSession.
If you are working from the sparkR shell, the SparkSession should already be created for you, and you would not need to call sparkR.session.
sparkR.session()
Starting Up from RStudio
You can also start SparkR from RStudio.
You can connect your R program to a Spark cluster from RStudio, R shell, Rscript or other R IDEs.
To start, make sure SPARK_HOME is set in environment (you can check Sys.getenv),
load the SparkR package, and call sparkR.session as below.
It will check for the Spark installation, and, if not found, it will be downloaded and cached automatically.
Alternatively, you can also run install.spark manually.
In addition to calling sparkR.session, you could also specify certain Spark driver properties.
Normally these Application properties and Runtime Environment cannot be set programmatically, as the driver JVM process would have been started, in this case SparkR takes care of this for you.
To set them, pass them as you would other configuration properties in the sparkConfig argument to sparkR.session().
if (nchar(Sys.getenv("SPARK_HOME")) < 1) {
Sys.setenv(SPARK_HOME = "/home/spark")
}
library(SparkR, lib.loc = c(file.path(Sys.getenv("SPARK_HOME"), "R", "lib")))
sparkR.session(master = "local[*]", sparkConfig = list(spark.driver.memory = "2g"))
The following Spark driver properties can be set in sparkConfig with sparkR.session from RStudio:
Property Name
Property group
spark-submit equivalent
spark.master
Application Properties
--master
spark.yarn.keytab
Application Properties
--keytab
spark.yarn.principal
Application Properties
--principal
spark.driver.memory
Application Properties
--driver-memory
spark.driver.extraClassPath
Runtime Environment
--driver-class-path
spark.driver.extraJavaOptions
Runtime Environment
--driver-java-options
spark.driver.extraLibraryPath
Runtime Environment
--driver-library-path
Creating SparkDataFrames
With a SparkSession, applications can create SparkDataFrames from a local R data frame, from a Hive table, or from other data sources.
From local data frames
The simplest way to create a data frame is to convert a local R data frame into a SparkDataFrame.
Specifically, we can use as.DataFrame or createDataFrame and pass in the local R data frame to create a SparkDataFrame.
As an example, the following creates a SparkDataFrame based using the faithful dataset from R.
df <- as.DataFrame(faithful)# Displays the first part of the SparkDataFramehead(df)## eruptions waiting##1 3.600 79##2 1.800 54##3 3.333 74
From Data Sources
SparkR supports operating on a variety of data sources through the SparkDataFrame interface.
This section describes the general methods for loading and saving data using Data Sources.
You can check the Spark SQL programming guide for more specific options that are available for the built-in data sources.
The general method for creating SparkDataFrames from data sources is read.df.
This method takes in the path for the file to load and the type of data source, and the currently active SparkSession will be used automatically.
SparkR supports reading JSON, CSV and Parquet files natively, and through packages available from sources like Third Party Projects, you can find data source connectors for popular file formats like Avro.
These packages can either be added by
specifying --packages with spark-submit or sparkR commands, or if initializing SparkSession with sparkPackages parameter when in an interactive R shell or from RStudio.
We can see how to use data sources using an example JSON input file.
Note that the file that is used here is not a typical JSON file.
Each line in the file must contain a separate, self-contained valid JSON object.
For more information, please see JSON Lines text format, also called newline-delimited JSON.
As a consequence, a regular multi-line JSON file will most often fail.
people <- read.df("./examples/src/main/resources/people.json","json")head(people)## age name##1 NA Michael##2 30 Andy##3 19 Justin# SparkR automatically infers the schema from the JSON file
printSchema(people)# root# |-- age: long (nullable = true)# |-- name: string (nullable = true)# Similarly, multiple files can be read with read.json
people <- read.json(c("./examples/src/main/resources/people.json","./examples/src/main/resources/people2.json"))
The data sources API natively supports CSV formatted input files.
For more information please refer to SparkR read.df API documentation.
The data sources API can also be used to save out SparkDataFrames into multiple file formats.
For example, we can save the SparkDataFrame from the previous example
to a Parquet file using write.df.
You can also create SparkDataFrames from Hive tables.
To do this we will need to create a SparkSession with Hive support which can access tables in the Hive MetaStore.
Note that Spark should have been built with Hive support and more details can be found in the SQL programming guide.
In SparkR, by default it will attempt to create a SparkSession with Hive support enabled (enableHiveSupport = TRUE).
sparkR.session()
sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")
sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")# Queries can be expressed in HiveQL.
results <- sql("FROM src SELECT key, value")# results is now a SparkDataFramehead(results)## key value## 1 238 val_238## 2 86 val_86## 3 311 val_311
SparkDataFrame Operations
SparkDataFrames support a number of functions to do structured data processing.
Here we include some basic examples and a complete list can be found in the API docs:
Selecting rows, columns
# Create the SparkDataFrame
df <- as.DataFrame(faithful)# Get basic information about the SparkDataFrame
df
## SparkDataFrame[eruptions:double, waiting:double]# Select only the "eruptions" columnhead(select(df, df$eruptions))## eruptions##1 3.600##2 1.800##3 3.333# You can also pass in column name as stringshead(select(df,"eruptions"))# Filter the SparkDataFrame to only retain rows with wait times shorter than 50 minshead(filter(df, df$waiting <50))## eruptions waiting##1 1.750 47##2 1.750 47##3 1.867 48
Grouping, Aggregation
SparkR data frames support a number of commonly used functions to aggregate data after grouping.
For example, we can compute a histogram of the waiting time in the faithful dataset as shown below
# We use the `n` operator to count the number of times each waiting time appearshead(summarize(groupBy(df, df$waiting), count = n(df$waiting)))## waiting count##1 70 4##2 67 1##3 69 2# We can also sort the output from the aggregation to get the most common waiting times
waiting_counts <- summarize(groupBy(df, df$waiting), count = n(df$waiting))head(arrange(waiting_counts, desc(waiting_counts$count)))## waiting count##1 78 15##2 83 14##3 81 13
In addition to standard aggregations, SparkR supports OLAP cube operators cube:
head(agg(cube(df,"cyl","disp","gear"), avg(df$mpg)))## cyl disp gear avg(mpg)##1 NA 140.8 4 22.8##2 4 75.7 4 30.4##3 8 400.0 3 19.2##4 8 318.0 3 15.5##5 NA 351.0 NA 15.8##6 NA 275.8 NA 16.3
SparkR also provides a number of functions that can directly applied to columns for data processing and during aggregation.
The example below shows the use of basic arithmetic functions.
# Convert waiting time from hours to seconds.# Note that we can assign this to a new column in the same SparkDataFrame
df$waiting_secs <- df$waiting *60head(df)## eruptions waiting waiting_secs##1 3.600 79 4740##2 1.800 54 3240##3 3.333 74 4440
Applying User-Defined Function
In SparkR, we support several kinds of User-Defined Functions:
Run a given function on a large dataset using dapply or dapplyCollect
dapply
Apply a function to each partition of a SparkDataFrame.
The function to be applied to each partition of the SparkDataFrame
and should have only one parameter, to which a data.frame corresponds to each partition will be passed.
The output of function should be a data.frame.
Schema specifies the row format of the resulting a SparkDataFrame.
It must match to data types of returned value.
# Convert waiting time from hours to seconds.# Note that we can apply UDF to DataFrame.
schema <- structType(structField("eruptions","double"), structField("waiting","double"),
structField("waiting_secs","double"))
df1 <- dapply(df,function(x){ x <-cbind(x, x$waiting *60)}, schema)head(collect(df1))## eruptions waiting waiting_secs##1 3.600 79 4740##2 1.800 54 3240##3 3.333 74 4440##4 2.283 62 3720##5 4.533 85 5100##6 2.883 55 3300
dapplyCollect
Like dapply, apply a function to each partition of a SparkDataFrame and collect the result back.
The output of function
should be a data.frame.
But, Schema is not required to be passed.
Note that dapplyCollect can fail if the output of UDF run on all the partition cannot be pulled to the driver and fit in driver memory.
# Convert waiting time from hours to seconds.# Note that we can apply UDF to DataFrame and return a R's data.frame
ldf <- dapplyCollect(
df,function(x){
x <-cbind(x,"waiting_secs"= x$waiting *60)})head(ldf,3)## eruptions waiting waiting_secs##1 3.600 79 4740##2 1.800 54 3240##3 3.333 74 4440
Run a given function on a large dataset grouping by input column(s) and using gapply or gapplyCollect
gapply
Apply a function to each group of a SparkDataFrame.
The function is to be applied to each group of the SparkDataFrame and should have only two parameters: grouping key and R data.frame corresponding to
that key.
The groups are chosen from SparkDataFrames column(s).
The output of function should be a data.frame.
Schema specifies the row format of the resulting SparkDataFrame.
It must represent R function’s output schema on the basis of Spark data types.
The column names of the returned data.frame are set by user.
# Determine six waiting times with the largest eruption time in minutes.
schema <- structType(structField("waiting","double"), structField("max_eruption","double"))
result <- gapply(
df,"waiting",function(key, x){
y <-data.frame(key,max(x$eruptions))},
schema)head(collect(arrange(result,"max_eruption", decreasing =TRUE)))## waiting max_eruption##1 64 5.100##2 69 5.067##3 71 5.033##4 87 5.000##5 63 4.933##6 89 4.900
gapplyCollect
Like gapply, applies a function to each partition of a SparkDataFrame and collect the result back to R data.frame.
The output of the function should be a data.frame.
But, the schema is not required to be passed.
Note that gapplyCollect can fail if the output of UDF run on all the partition cannot be pulled to the driver and fit in driver memory.
# Determine six waiting times with the largest eruption time in minutes.
result <- gapplyCollect(
df,"waiting",function(key, x){
y <-data.frame(key,max(x$eruptions))colnames(y)<-c("waiting","max_eruption")
y
})head(result[order(result$max_eruption, decreasing =TRUE),])## waiting max_eruption##1 64 5.100##2 69 5.067##3 71 5.033##4 87 5.000##5 63 4.933##6 89 4.900
Run local R functions distributed using spark.lapply
spark.lapply
Similar to lapply in native R, spark.lapply runs a function over a list of elements and distributes the computations with Spark.
Applies a function in a manner that is similar to doParallel or lapply to elements of a list.
The results of all the computations should fit in a single machine.
If that is not the case they can do something like df <- createDataFrame(list) and then use dapply
# Perform distributed training of multiple models with spark.lapply.
Here, we pass# a read-only list of arguments which specifies family the generalized linear model should be.
families <-c("gaussian","poisson")
train <-function(family){
model <- glm(Sepal.Length ~ Sepal.Width + Species, iris, family = family)summary(model)}# Return a list of model's summaries
model.summaries <- spark.lapply(families, train)# Print the summary of each modelprint(model.summaries)
Running SQL Queries from SparkR
A SparkDataFrame can also be registered as a temporary view in Spark SQL and that allows you to run SQL queries over its data.
The sql function enables applications to run SQL queries programmatically and returns the result as a SparkDataFrame.
# Load a JSON file
people <- read.df("./examples/src/main/resources/people.json","json")# Register this SparkDataFrame as a temporary view.
createOrReplaceTempView(people,"people")# SQL statements can be run by using the sql method
teenagers <- sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")head(teenagers)## name##1 Justin
Machine Learning
Algorithms
SparkR supports the following machine learning algorithms currently:
spark.kstest: Kolmogorov-Smirnov Test
Under the hood, SparkR uses MLlib to train the model.
Please refer to the corresponding section of MLlib user guide for example code.
Users can call summary to print a summary of the fitted model, predict to make predictions on new data, and write.ml/read.ml to save/load fitted models.
SparkR supports a subset of the available R formula operators for model fitting, including ‘~’, ‘.’, ‘:’, ‘+’, and ‘-‘.
Model persistence
The following example shows how to save/load a MLlib model by SparkR.
training <- read.df("data/mllib/sample_multiclass_classification_data.txt",source="libsvm")# Fit a generalized linear model of family "gaussian" with spark.glm
df_list <- randomSplit(training,c(7,3),2)
gaussianDF <- df_list[[1]]
gaussianTestDF <- df_list[[2]]
gaussianGLM <- spark.glm(gaussianDF, label ~ features, family ="gaussian")# Save and then load a fitted MLlib model
modelPath <-tempfile(pattern ="ml", fileext =".tmp")
write.ml(gaussianGLM, modelPath)
gaussianGLM2 <- read.ml(modelPath)# Check model summarysummary(gaussianGLM2)# Check model prediction
gaussianPredictions <- predict(gaussianGLM2, gaussianTestDF)head(gaussianPredictions)unlink(modelPath)
Find full example code at "examples/src/main/r/ml/ml.R" in the Spark repo.
SparkR supports the Structured Streaming API.
Structured Streaming is a scalable and fault-tolerant stream processing engine built on the Spark SQL engine.
For more information see the R API on the Structured Streaming Programming Guide
R Function Name Conflicts
When loading and attaching a new package in R, it is possible to have a name conflict, where a
function is masking another function.
The following functions are masked by the SparkR package:
Masked function
How to Access
cov in package:stats
stats::cov(x, y = NULL, use = "everything",
method = c("pearson", "kendall", "spearman"))
Since part of SparkR is modeled on the dplyr package, certain functions in SparkR share the same names with those in dplyr.
Depending on the load order of the two packages, some functions from the package loaded first are masked by those in the package loaded after.
In such case, prefix such calls with the package name, for instance, SparkR::cume_dist(x) or dplyr::cume_dist(x).
You can inspect the search path in R with search()
https://spark.apache.org/docs/latest/sparkr.html
Once you've connected to Spark you'll be able to browse the tables contained within the Spark cluster and preview Spark DataFrames using the standard RStudio data viewer:
You can also connect to Spark through Livy through a new connection dialog:
The RStudio IDE features for sparklyr are available now as part of the RStudio Preview Release.