Building Spark ML pipelines with sparklyr David Wilson | March 29, 2026 Building Spark ML pipelines with sparklyr class: center, middle, inverse, title-slide # Building Spark ML pipelines<br>with sparklyr ### Kevin Kuo ### February 2018 --- class: rstudio-slide, left # Quick Recap **sparklyr** is an R package that provides an interface to Apache Spark, an in-memory, cluster-computing framework. -- - A dplyr interface for manipulating data. -- - Machine learning capabilities. -- - Ecosystem of extensions. -- - Spelled **sparklyr**, not **SparklyR**. --- class: rstudio-slide, left # sparklyr 0.7 Hit CRAN last week! 🎉 - **ML Pipelines**: A high-level interface for building ML workflows in Spark -- - Full support for feature transformers and machine learning algorithms - Data serialization improvements - YARN cluster mode support - Various other enhancements, see **blog.rstudio.com** and **spark.rstudio.com** --- class: dark-gray, middle # ML Pipelines --- class: rstudio-slide, left # Why are pipelines cool? - Easy organization of complex data transformation and ML workflows. -- - Automated tuning of models and pipelines. -- - Model persistence for deployment. --- class: rstudio-slide, left # Jargon - A `Transformer` takes a data frame, via `ml_transform()`, and returns a transformed data frame.  --- class: rstudio-slide, left # Jargon - A `Transformer` takes a data frame, via `ml_transform()`, and returns a transformed data frame. - An `Estimator` take a data frame, via `ml_fit()`, and returns a `Transformer`.  --- class: rstudio-slide, left # Jargon - A `Transformer` takes a data frame, via `ml_transform()`, and returns a transformed data frame. - An `Estimator` take a data frame, via `ml_fit()`, and returns a `Transformer`. - A `Pipeline` consists of a sequence of stages—`PipelineStage`s—that act on some data in order. - A `PipelineStage` can be either a `Transformer` or an `Estimator`.  --- class: rstudio-slide, left # Jargon - A `Pipeline` consists of a sequence of stages—`PipelineStage`s—that act on some data in order. - A `PipelineStage` can be either a `Transformer` or an `Estimator`. - A `Pipeline` is always an `Estimator`, and its fitted form is called `PipelineModel` which is a `Transformer`.  --- class: rstudio-slide, left # Jargon in pictures  --- class: rstudio-slide, left # Jargon in pictures  --- class: dark-gray, middle # Let's see some code! --- class: rstudio-slide, left, sparklyr-code # Small data, because we're humble We gonna classify some flowers! r library(sparklyr) sc <- spark_connect(master = "local") # copy data to Spark and partition into train/validation iris_tbls <- sdf_copy_to(sc, iris, overwrite = TRUE) %>% sdf_partition(train = 2/3, validation = 1/3, seed = 2018) train <- iris_tbls$train validation <- iris_tbls$validation --- class: rstudio-slide, left, sparklyr-code20 # Fitting a quick model r model <- ml_logistic_regression(train, Species ~ .) model ## Call: ml_logistic_regression.tbl_spark(train, Species ~ .) ## ## Formula: Species ~ . ## ## Coefficients: ## (Intercept) Sepal_Length Sepal_Width Petal_Length Petal_Width ## virginica -16.6751258 3.479078 -19.12870 7.81782098 16.750741 ## versicolor 16.2061082 5.520549 -12.57257 0.07125575 1.121679 ## setosa 0.4690176 -8.999627 31.70126 -7.88907673 -17.872420 --- class: rstudio-slide, left, sparklyr-code # Predictions! r library(dplyr) ml_predict(model, validation) %>% # sdf_predict(validation, model) select(Species, label, prediction, predicted_label) %>% head(7) ## # Source: lazy query [?? x 4] ## # Database: spark_connection ## Species label prediction predicted_label ## <chr> <dbl> <dbl> <chr> ## 1 setosa 2.00 2.00 setosa ## 2 setosa 2.00 2.00 setosa ## 3 setosa 2.00 2.00 setosa ## 4 setosa 2.00 2.00 setosa ## 5 setosa 2.00 2.00 setosa ## 6 setosa 2.00 2.00 setosa ## 7 versicolor 1.00 1.00 versicolor --- class: rstudio-slide, left, sparklyr-code # The deets `model` actually contains a pipeline model! Let's go through an example implementation. --- class: rstudio-slide, left # Spark peculiarities Spark ML - Requires that all **predictors** be compressed into a **single vector column**. - Requires that the **response** column be **numeric**, even for classification problems. --- class: rstudio-slide, left, sparklyr-code20 # Vector assembler - A transformer that concatenates columns into a vector column. r vector_assembler <- ft_vector_assembler( sc, input_cols = c("Sepal_Length", "Sepal_Width", "Petal_Length", "Petal_Width"), output_col = "features" ) vector_assembler ## VectorAssembler (Transformer) ## <vector_assembler_1674160c373ee> ## (Parameters -- Column Names) ## input_cols: Sepal_Length, Sepal_Width, Petal_Length, Petal_Width ## output_col: features --- class: rstudio-slide, left, sparklyr-code20 # Vector assembler in action r vector_assembler %>% ml_transform(train) %>% glimpse() ## Observations: ?? ## Variables: 6 ## $ Sepal_Length <dbl> 4.3, 4.4, 4.4, 4.6, 4.6, 4.6, 4.6, 4.7, 4.7, 4.8,... ## $ Sepal_Width <dbl> 3.0, 2.9, 3.0, 3.1, 3.2, 3.4, 3.6, 3.2, 3.2, 3.0,... ## $ Petal_Length <dbl> 1.1, 1.4, 1.3, 1.5, 1.4, 1.4, 1.0, 1.3, 1.6, 1.4,... ## $ Petal_Width <dbl> 0.1, 0.2, 0.2, 0.2, 0.2, 0.3, 0.2, 0.2, 0.2, 0.1,... ## $ Species <chr> "setosa", "setosa", "setosa", "setosa", "setosa",... ## $ features <list> [<4.3, 3.0, 1.1, 0.1>, <4.4, 2.9, 1.4, 0.2>, <4.... --- class: rstudio-slide, left # String indexer - An estimator to encode the `Species` column into a numeric column. r string_indexer <- ft_string_indexer(sc, "Species", "label") string_indexer_model <- string_indexer %>% ml_fit(train) string_indexer_model %>% ml_transform(train) %>% select(Species, label) %>% head(4) ## # Source: lazy query [?? x 2] ## # Database: spark_connection ## Species label ## <chr> <dbl> ## 1 setosa 2.00 ## 2 setosa 2.00 ## 3 setosa 2.00 ## 4 setosa 2.00 --- class: rstudio-slide, left # String indexer - label mapping - Keeping track of the label-index mapping - Note 0-based indexing r labels <- ml_labels(string_indexer_model) labels ## [1] "virginica" "versicolor" "setosa" --- class: rstudio-slide, left # A pipeline! <style type="text/css"> .{.foobar}.hljs.vala.remark-code { height: 220px; overflow: scroll; } .{.foobar}.hljs.vala.remark-code span { font-weight:normal; color:#000; } </style> r pipeline <- ml_pipeline( vector_assembler, string_indexer_model ) %>% ml_logistic_regression() %>% ft_index_to_string("prediction", "predicted_label", labels = labels) pipeline {.foobar} ## Pipeline (Estimator) with 4 stages ## <pipeline_16741761531f6> ## Stages ## |--1 VectorAssembler (Transformer) ## | <vector_assembler_1674160c373ee> ## | (Parameters -- Column Names) ## | input_cols: Sepal_Length, Sepal_Width, Petal_Length, Petal_Width ## | output_col: features ## |--2 StringIndexerModel (Transformer) ## | <string_indexer_16741773b6bfe> ## | (Parameters -- Column Names) ## | input_col: Species ## | output_col: label ## | (Transformer Info) ## | labels: chr [1:3] "virginica" "versicolor" "setosa" ## |--3 LogisticRegression (Estimator) ## | <logistic_regression_167412a720ca4> ## | (Parameters -- Column Names) ## | features_col: features ## | label_col: label ## | prediction_col: prediction ## | probability_col: probability ## | raw_prediction_col: rawPrediction ## | (Parameters) ## | aggregation_depth: 2 ## | elastic_net_param: 0 ## | family: auto ## | fit_intercept: TRUE ## | max_iter: 100 ## | reg_param: 0 ## | standardization: TRUE ## | threshold: 0.5 ## | tol: 1e-06 ## |--4 IndexToString (Transformer) ## | <index_to_string_1674122e85a31> ## | (Parameters -- Column Names) ## | input_col: prediction ## | output_col: predicted_label --- class: rstudio-slide, left # Fitting the pipeline r pipeline_model <- pipeline %>% ml_fit(train) pipeline_model %>% ml_transform(validation) %>% glimpse() ## Observations: ?? ## Variables: 11 ## $ Sepal_Length <dbl> 4.4, 4.5, 4.8, 4.8, 4.9, 4.9, 5.0, 5.0, 5.0, 5... ## $ Sepal_Width <dbl> 3.2, 2.3, 3.0, 3.4, 3.0, 3.1, 2.0, 3.3, 3.4, 3... ## $ Petal_Length <dbl> 1.3, 1.3, 1.4, 1.6, 1.4, 1.5, 3.5, 1.4, 1.5, 1... ## $ Petal_Width <dbl> 0.2, 0.3, 0.3, 0.2, 0.2, 0.1, 1.0, 0.2, 0.2, 0... ## $ Species <chr> "setosa", "setosa", "setosa", "setosa", "setos... ## $ features <list> [<4.4, 3.2, 1.3, 0.2>, <4.5, 2.3, 1.3, 0.3>, ... ## $ label <dbl> 2, 2, 2, 2, 2, 2, 1, 2, 2, 2, 2, 2, 2, 2, 1, 2... ## $ rawPrediction <list> [<-49.0656919, 0.5812787, 48.4844132>, <-29.8... ## $ probability <list> [<4.310500e-43, 1.570121e-21, 1.000000e+00>, ... ## $ prediction <dbl> 2, 2, 2, 2, 2, 2, 1, 2, 2, 2, 2, 2, 2, 2, 1, 2... ## $ predicted_label <chr> "setosa", "setosa", "setosa", "setosa", "setos... --- class: dark-gray, middle # On deployment and collaboration --- class: rstudio-slide, sparklyr-code # If you like it, keep it r ml_save(pipeline_model, "path/to/model") ...to get it back, r pipeline_model <- ml_load(sc, "path/to/model") predictions <- pipeline_model %>% ml_transform(new_data) --- class: rstudio-slide, sparklyr-code # You can save the unfitted pipeline, too r ml_save(pipeline, "path/to/pipeline") ...refresh the model with new data: r pipeline <- ml_load(sc, "path/to/pipeline") new_pipeline_model <- pipeline %>% ml_fit(new_data) --- class: rstudio-slide, left # Easy deployment - Reports on a schedule - Automatic model refreshes - Batch predictions over an API  --- class: rstudio-slide, left # We also play nice with others .pull-left[ - Saved pipelines are fully interoperable  ] .pull-right[ scala import org.apache.spark.ml.{Pipeline, PipelineModel} val pipeline = Pipeline.load("path/to/pipeline") val model = PipelineModel.load("path/to/model") ] --- class: dark-gray, middle # Other things to try at home/work --- class: rstudio-slide, left # Model tuning - Cross validation r # specify hyperparameter grid param_grid <- list( logistic = list( elastic_net_param = list(0, 0.01), reg_param = list(0, 0.01) )) # create cross validator estimator and fit cv <- ml_cross_validator( sc, ml_logistic_regression(sc), param_grid, evaluator = ml_multiclass_classification_evaluator(sc)) cvm <- ml_fit(cv, ft_r_formula(train, Species ~ .)) --- class: rstudio-slide, left # Dplyr transformer - Defining a transformer via **dplyr** transformations r library(dplyr) transform_tbl <- train %>% group_by(Species) %>% count() sql_transformer <- ft_dplyr_transformer(sc, transform_tbl) sql_transformer %>% ml_transform(validation) ## # Source: table<sparklyr_tmp_167415050a376> [?? x 2] ## # Database: spark_connection ## Species n ## <chr> <dbl> ## 1 versicolor 18.0 ## 2 virginica 13.0 ## 3 setosa 21.0 --- class: rstudio-slide, left # Pipeline architecture  --- class: dark-gray, middle # Needs halp?! --- class: rstudio-slide, left # Resources - Docs — spark.rstudio.com - GitHub repo — rstudio/sparklyr - StackOverflow - Gitter channel — - Tidyverse/IDE/Markdown — community.rstudio.com --- class: dark-gray, middle # Thank you!
