We are excited to share that sparklyr 0.7 is now available on CRAN! Sparklyr provides an R interface to Apache Spark. It supports dplyr syntax for working with Spark DataFrames and exposes the full range of machine learning algorithms available in Spark. You can also learn more about Apache Spark and sparklyr in spark.rstudio.com and our new webinar series on Apache Spark. Features in this release:
Adds support for ML Pipelines which provide a uniform set of high-level APIs to help create, tune, and deploy machine learning pipelines at scale.
Enhances Machine Learning capabilities by supporting the full range of ML algorithms and feature transformers.
Improves Data Serialization, specifically by adding support for date columns.
Adds support for YARN cluster mode connections.
Adds various other improvements as listed in the NEWS file.
In this blog post, we highlight Pipelines, new ML functions, and enhanced support for data serialization. To follow along in the examples below, you can upgrade to the latest stable version from CRAN with:
install.packages(&quot;sparklyr&quot;)
ML Pipelines
The ML Pipelines API is a high-level interface for building ML workflows in Spark. Pipelines provide a uniform approach to compose feature transformers and ML routines, and are interoperable across the different Spark APIs (R/sparklyr, Scala, and Python.)
First, let’s go over a quick overview of terminology. A Pipeline consists of a sequence of stages—PipelineStages—that act on some data in order. A PipelineStage can be either a Transformer or an Estimator. A Transformer takes a data frame and returns a transformed data frame, whereas an Estimator take a data frame and returns a Transformer. You can think of an Estimator as an algorithm that can be fit to some data, e.g. the ordinary least squares (OLS) method, and a Transformer as the fitted model, e.g. the linear formula that results from OLS. A Pipeline is itself a PipelineStage and can be an element in another Pipeline. Lastly, a Pipeline is always an Estimator, and its fitted form is called PipelineModel which is a Transformer.
Let’s look at some examples of creating pipelines. We establish a connection and copy some data to Spark:
library(sparklyr)
library(dplyr)
# If needed, install Spark locally via `spark_install()`
sc %
mutate(Sepal_Length = log(Sepal_Length),
Sepal_Width = Sepal_Width ^ 2)
) %__%
ft_string_indexer(&quot;Species&quot;, &quot;label&quot;)
pipeline
## Pipeline (Estimator) with 2 stages
## ## Stages ## |--1 SQLTransformer (Transformer)
## | ## | (Parameters -- Column Names)
## |--2 StringIndexer (Estimator)
## | ## | (Parameters -- Column Names)
## | input_col: Species
## | output_col: label
## | (Parameters)
## | handle_invalid: error
Under the hood, ft_dplyr_transformer() extracts the SQL statements associated with the input and creates a Spark SQLTransformer, which can then be applied to new datasets with the appropriate columns. We now fit the Pipeline with ml_fit() then transform some data using the resulting PipelineModel with ml_transform().
pipeline_model %
ml_fit(iris_data$train)
# pipeline_model is a transformer
pipeline_model %__%
ml_transform(iris_data$validation) %__%
glimpse()
## Observations: ??
## Variables: 6
## $ Petal_Length 1.4, 1.3, 1.3, 1.0, 1.6, 1.9, 3.3, 4.5, 1.6, 1.5,...
## $ Petal_Width 0.2, 0.2, 0.2, 0.2, 0.2, 0.2, 1.0, 1.7, 0.2, 0.2,...
## $ Species &quot;setosa&quot;, &quot;setosa&quot;, &quot;setosa&quot;, &quot;setosa&quot;, &quot;setosa&quot;,...
## $ Sepal_Length 1.482, 1.482, 1.482, 1.526, 1.548, 1.569, 1.589, ...
## $ Sepal_Width 8.41, 9.00, 10.24, 12.96, 10.24, 11.56, 5.76, 6.2...
## $ label 1, 1, 1, 1, 1, 1, 0, 2, 1, 1, 1, 0, 1, 1, 1, 1, 1...
A predictive modeling pipeline
Now, let’s try to build a classification pipeline on the iris dataset.
Spark ML algorithms require that the label column be encoded as numeric and predictor columns be encoded as one vector column. We’ll build on the pipeline we created in the previous section, where we have already included a StringIndexer stage to encode the label column.
# define stages
# vector_assember will concatenate the predictor columns into one vector column
vector_assembler %
collect() %__%
mutate_all(function(x) sapply(x, paste0, collapse = &quot;,&quot;))
## # A tibble: 3 x 2
## items prediction
## ## 1 1,2,5 &quot;&quot; ## 2 1,2,3,5 &quot;&quot; ## 3 1,2 5
Data serialization
Various improvements were made to better support serialization and collection of data frames. Most notably, dates are now supported:
copy_to(sc, nycflights13::flights) %__%
select(carrier, flight, time_hour)
## # Source: lazy query [?? x 3]
## # Database: spark_connection
## carrier flight time_hour ## ## 1 UA 1545 2013-01-01 05:00:00
## 2 UA 1714 2013-01-01 05:00:00
## 3 AA 1141 2013-01-01 05:00:00
## 4 B6 725 2013-01-01 05:00:00
## 5 DL 461 2013-01-01 06:00:00
## 6 UA 1696 2013-01-01 05:00:00
## 7 B6 507 2013-01-01 06:00:00
## 8 EV 5708 2013-01-01 06:00:00
## 9 B6 79 2013-01-01 06:00:00
## 10 AA 301 2013-01-01 06:00:00
## # ... with more rows
We can’t wait to see what you’ll build with the new features! As always, comments, issue reports, and contributions are welcome on the sparklyr GitHub repo.