<p><div><div><div><div data-show-faces="false" data-height="21" data-width="100" data-layout="button_count" data-send="true" data-href="https://www.r-bloggers.com/leveraging-pipeline-in-spark-trough-scala-and-sparklyr/"></div></div><div></div></div><div></div></div>
<div>
(This article was first published on <strong><a href="http://r-posts.com/leveraging-pipeline-in-spark-trough-scala-and-sparklyr/"> R-posts.com</a></strong>, and kindly contributed to <a rel="nofollow" href="https://www.r-bloggers.com/">R-bloggers)</a> </div>
<h2>Intro</h2>
<p>Pipeline concept is definitely not new for software world, Unix pipe operator (|) links two tasks putting the output of one as the input of the other. In machine learning solutions it is pretty much usual to apply several transformation and manipulation to datasets, or to different portions or sample of the same dataset (from classic test/train slices to samples taken for cross-validation procedure). In these cases, pipelines are definitely useful to define a sequence of tasks chained together to define a complete process, which in turn simplifies the operation of the ml solution. In addition, in BigData environment, it is possible to apply the “laziness” of execution to the entire process in order to make it more scalable and robust, therefore no surprise to see pipeline implemented in Spark machine learning library and R API available, by SparklyR package, to leverage the construct. Pipeline component in Spark are basically of two types :</p>
<ul>
<li>
<strong>transformer:</strong> since dataframe usually need to undergo several kinds of changes column-wide, row-wide or even single value-wide, transformers are the component meant to deliver these transformations. Typically a transformer has a table or dataframe as input and delivers a table/dataframe as output. Sparks, through MLlib, provide a<span> </span><a rel="nofollow" href="http://spark.apache.org/docs/latest/ml-features.html" target="_blank">set of feature’s transformers</a><span> </span>meant to address most common transformations needed;
</li>
<li>
<strong>estimator:</strong><span></span> estimator is the component which delivers a<span> </span>model,<span></span> fitting an algorithm to train data. Indeed<span> </span><em>fit()</em><span> </span>is the key method for an estimator and produces, as said a<span> </span><em>model</em><span> </span>which is a transformer. Leveraging the parallel processing which is provided by Spark, it is possible to run several estimators in parallel on different training dataset in order to find the best solution (or even to avoid overfitting issue).<span> </span><em>ML algorithms</em><span> </span>are basically a set of Estimators, they build a rich set of machine learning (ML) common algorithms, available from MLlib. This is a library of algorithms meant to be scalable and run in a parallel environment. Starting from the 2.0 release of Spark, the RDD-based library is in maintenance mode (the RDD-based APIs are expected to be removed in 3.0 release) whereas the mainstream development is focused on supporting dataframes. In MLlib features are to be expressed with<span> </span><em>labeledpoints</em>, which means numeric vectors for features and predictors.Pipelines of transformers are, even for this reason, extremely useful to operationalize an ML solutions Spark-based. For additional details on MLlib refer to<span> </span><a rel="nofollow" href="https://spark.apache.org/docs/latest/ml-guide.html" target="_blank">Apache Spark documentation</a>
</li>
</ul>
<p>In this post we’ll see a simple example of pipeline usage and we’ll see two version of the same example: the first one using Scala (which is a kind of “native” language for Spark environment), afterward we’ll see how to implement the same example in R, leveraging SaprklyR package in order to show how powerful and complete it is.</p>
<div>
<h3>The dataset</h3>
<p>For this example, the dataset comes from<span> </span><a rel="nofollow" href="http://archive.ics.uci.edu/ml/index.php" target="_blank">UCI – Machine Learning Repository Irvine, CA: University of California, School of Information and Computer Science</a>. “Adults Income” dataset reports individual’s annual income results from various factors. The annual income will be our label (it is divided into two classes: &lt;=50K and >50K) and there are 14 features, for each individual, we can leverage to explore the possibility in predicting income level. For additional detail on “adults dataset” refer to the UCI machine learning repository<span> </span><a rel="nofollow" href="http://www.cs.toronto.edu/~delve/data/adult/desc.html" target="_blank">http://www.cs.toronto.edu/~delve/data/adult/desc.html</a>.
</div>
<div>
<h3>Scala code</h3>
<p>As said, we’ll show how we can use scala API to access pipeline in MLlib, therefore we need to include references to classes we’re planning to use in our example and to start a spark session :</p>
<pre><code><span>import</span> org.<span>apache</span>.<span>spark</span>.<span>sql</span>.<span>types</span>._
<span>import</span> org.<span>apache</span>.<span>spark</span>.<span>sql</span>.<span>functions</span>._
<span>import</span> org.<span>apache</span>.<span>spark</span>.<span>sql</span>._

<span>import</span> org.<span>apache</span>.<span>spark</span>.<span>ml</span>.<span>Pipeline</span>
<span>import</span> org.<span>apache</span>.<span>spark</span>.<span>ml</span>.<span>feature</span>.{VectorAssembler, StringIndexer, VectorIndexer}
<span>import</span> org.<span>apache</span>.<span>spark</span>.<span>ml</span>.<span>classification</span>.<span>LogisticRegression</span>
<span>import</span> org.<span>apache</span>.<span>spark</span>.<span>sql</span>.<span>SparkSession</span></code></pre>
<div>
then we’ll read dataset and will start to manipulate data in order to prepare for the pipeline. In our example, we’ll get data out of local repository (instead of referring to an eg. HDFS or Datalake repository, there are API – for both scala and R – which allows the access to these repositories as well). We’ll leverage this upload activity also to rename some columns, in particular, we’ll rename the “income” column as “label” since we’ll use this a label column in our logistic regression algorithm.
<div>
<pre><code>
<span>//load data source from local repository</span>
<span>val</span> csv = spark.<span>read</span>.<span>option</span>(<span>&quot;inferSchema&quot;</span>,<span>&quot;true&quot;</span>)
  .<span>option</span>(<span>&quot;header&quot;</span>, <span>&quot;true&quot;</span>).<span>csv</span>(<span>&quot;...\yyyy\xxxx</span><span>\a</span><span>dult.csv&quot;</span>)

<span>val</span> data_start = csv.<span>select</span>(($<span>&quot;workclass&quot;</span>),($<span>&quot;gender&quot;</span>),($<span>&quot;education&quot;</span>),($<span>&quot;age&quot;</span>),
($<span>&quot;marital-status&quot;</span>).<span>alias</span>(<span>&quot;marital&quot;</span>), ($<span>&quot;occupation&quot;</span>),($<span>&quot;relationship&quot;</span>), 
($<span>&quot;race&quot;</span>),($<span>&quot;hours-per-week&quot;</span>).<span>alias</span>(<span>&quot;hr_per_week&quot;</span>), ($<span>&quot;native-country&quot;</span>).<span>alias</span>(<span>&quot;country&quot;</span>),
($<span>&quot;income&quot;</span>).<span>alias</span>(<span>&quot;label&quot;</span>),($<span>&quot;capital-gain&quot;</span>).<span>alias</span>(<span>&quot;capitalgain&quot;</span>),
($<span>&quot;capital-loss&quot;</span>).<span>alias</span>(<span>&quot;capitalloss&quot;</span>),($<span>&quot;educational-num&quot;</span>).<span>alias</span>(<span>&quot;eduyears&quot;</span>)).<span>toDF</span></code></pre>
</div>
<p>We’ll do some data clean up basically recoding the “working class” and “marital” columns, in order to reduce the number of codes and we’ll get rid of rows for which “occupation”, “working class”” (even recoded) and “capital gain” are not available. For first two column the dataset has the “?” value instead of “NA”, for capital gain there’s the 99999 value which will be filtered out. To recode “working class” and “marital” columns we’ll use UDF functions which in turn are wrappers of the actual recoding functions. To add a new column to the (new) dataframe we’ll use the “withColumn” method which will add “new_marital” and “new_workclass” to the<span> </span><em>startingdata</em><span> </span>dataframe. Afterwards, we’ll filter out all missing values and we’ll be ready to build the pipeline.</p>
<div>
<pre><code>
<span>// recoding marital status and working class, adding a new column </span>
<span>def</span> <span>newcol_marital</span>(str1:String): String = { 
    <span>var</span> nw_str = <span>&quot;noVal&quot;</span>
     <span>if</span> ((str1 == <span>&quot;Married-spouse-absent&quot;</span>) | (str1 ==<span>&quot;Married-AF-spouse&quot;</span>) 
         | (str1 == <span>&quot;Married-civ-spouse&quot;</span>)) {nw_str = <span>&quot;Married&quot;</span> } 
        <span>else</span> <span>if</span> ((str1 == <span>&quot;Divorced&quot;</span>) | (str1 == <span>&quot;Never-married&quot;</span> ) 
                 | (str1 == <span>&quot;Separated&quot;</span> ) | (str1 == <span>&quot;Widowed&quot;</span>)) 
          {nw_str = <span>&quot;Nonmarried&quot;</span> } 
        <span>else</span> { nw_str = str1}
    <span>return</span> nw_str
}
<span>val</span> udfnewcol = <span>udf</span>(newcol_marital _)
<span>val</span> recodeddata = data_start.<span>withColumn</span>(<span>&quot;new_marital&quot;</span>, <span>udfnewcol</span>('marital'))

<span>def</span> <span>newcol_wkclass</span>(str1:String): String = { 
    <span>var</span> nw_str = <span>&quot;noVal&quot;</span>
     <span>if</span> ((str1 == <span>&quot;Local-gov&quot;</span>) | (str1 ==<span>&quot;Federal-gov&quot;</span>) | (str1 == <span>&quot;State-gov&quot;</span>)) 
        {nw_str = <span>&quot;Gov&quot;</span> } 
    <span>else</span> <span>if</span> ((str1 == <span>&quot;Self-emp-not-inc&quot;</span>) | (str1 == <span>&quot;Self-emp-inc&quot;</span> )) 
        {nw_str = <span>&quot;Selfemployed&quot;</span> } 
    <span>else</span> <span>if</span> ((str1 == <span>&quot;Never-worked&quot;</span>) | (str1 == <span>&quot;Without-pay&quot;</span>)) 
        {nw_str = <span>&quot;Notemployed&quot;</span> } 
    <span>else</span> { nw_str = str1}
    <span>return</span> nw_str
}

<span>val</span> udfnewcol = <span>udf</span>(newcol_wkclass _)
<span>val</span> startingdata = recodeddata.<span>withColumn</span>(<span>&quot;new_workclass&quot;</span>, <span>udfnewcol</span>('workclass'))

<span>// remove missing data</span>
<span>val</span> df_work01 = startingdata.<span>na</span>.<span>drop</span>(<span>&quot;any&quot;</span>)
<span>val</span> df_work = startingdata.<span>filter</span>(<span>&quot;occupation &lt;>'?' </span>
                                and capitalgain &lt; <span>99999</span> 
                                and new_workclass &lt;> <span>'?'</span> 
                                and country &lt;> <span>'?'</span> <span>&quot;)</span></code></pre>
</div>
<p>In our example, we’re going to use 12 features, 7 are categorical variables and 5 are numeric variables. The feature’s array we’ll use to fit the model will be the results of merging two arrays, one for categorical variables and the second one for numeric variables. Before building the categorical variables array, we need to transform categories to indexes using transformers provided by spark.ml, even the<span> </span><em>label</em><span> </span>must be transformed to an index. Our pipeline then will include 7 pipeline<span> </span><em>stages</em><span> </span>to transform categorical variables, 1<span> </span><em>stage</em><span> </span>to transform the label, 2<span> </span><em>stages</em><span> </span>to build categorical and numeric arrays, and the final<span> </span><em>stage</em><span> </span>to fit the logistic model. Finally, we’ll build an 11-stages pipeline.<br>
To transform categorical variables into index we’ll use “Stringindexer”<span> </span><em>Transformer</em>. StringIndexer encodes a vector of string to a column of non-negative indices, ranging from 0 to the number of values. The indices ordered by label frequencies, so the most frequent value gets index 0. For each variable, we need to define the<span> </span><em>input column</em><span> </span>and the<span> </span><em>output column</em><span> </span>which we’ll use as input for other transformer or evaluators. Finally it is possible to define the strategy to handle unseen labels (possible when you use the S<em>tringindexer</em> to fit a model and run the fitted model against a test dataframe) through<span> </span><em>setHandleInvalid</em><span> </span>method , in our case we simply put “skip” to tell<span> </span><em>Stringindexer</em><span> </span>we want to skip unseen labels (additional details are available in MLlib documentation).</p>
<div>
<pre><code>
<span>// define stages</span>
<span>val</span> new_workclassIdx = <span>new</span> <span>StringIndexer</span>().<span>setInputCol</span>(<span>&quot;new_workclass&quot;</span>)
.<span>setOutputCol</span>(<span>&quot;new_workclassIdx&quot;</span>).<span>setHandleInvalid</span>(<span>&quot;skip&quot;</span>)

<span>val</span> genderIdx = <span>new</span> <span>StringIndexer</span>().<span>setInputCol</span>(<span>&quot;gender&quot;</span>)
.<span>setOutputCol</span>(<span>&quot;genderIdx&quot;</span>).<span>setHandleInvalid</span>(<span>&quot;skip&quot;</span>)

<span>val</span> maritalIdx = <span>new</span> <span>StringIndexer</span>().<span>setInputCol</span>(<span>&quot;new_marital&quot;</span>)
.<span>setOutputCol</span>(<span>&quot;maritalIdx&quot;</span>).<span>setHandleInvalid</span>(<span>&quot;skip&quot;</span>)

<span>val</span> occupationIdx = <span>new</span> <span>StringIndexer</span>().<span>setInputCol</span>(<span>&quot;occupation&quot;</span>)
.<span>setOutputCol</span>(<span>&quot;occupationIdx&quot;</span>).<span>setHandleInvalid</span>(<span>&quot;skip&quot;</span>)

<span>val</span> relationshipIdx = <span>new</span> <span>StringIndexer</span>().<span>setInputCol</span>(<span>&quot;relationship&quot;</span>)
.<span>setOutputCol</span>(<span>&quot;relationshipIdx&quot;</span>).<span>setHandleInvalid</span>(<span>&quot;skip&quot;</span>)

<span>val</span> raceIdx = <span>new</span> <span>StringIndexer</span>().<span>setInputCol</span>(<span>&quot;race&quot;</span>)
.<span>setOutputCol</span>(<span>&quot;raceIdx&quot;</span>).<span>setHandleInvalid</span>(<span>&quot;skip&quot;</span>)

<span>val</span> countryIdx = <span>new</span> <span>StringIndexer</span>().<span>setInputCol</span>(<span>&quot;country&quot;</span>)
.<span>setOutputCol</span>(<span>&quot;countryIdx&quot;</span>).<span>setHandleInvalid</span>(<span>&quot;skip&quot;</span>)

<span>val</span> labelIdx = <span>new</span> <span>StringIndexer</span>().<span>setInputCol</span>(<span>&quot;label&quot;</span>)
.<span>setOutputCol</span>(<span>&quot;labelIdx&quot;</span>).<span>setHandleInvalid</span>(<span>&quot;skip&quot;</span>)</code></pre>
</div>
<p>In addition to<span> </span><em>Transfomer</em><span> </span>and<span> </span><em>Estimator</em><span> </span>provided by spark.ml package, it is possible to define custom Estimator and Transformers. As an example we’ll see how to define a custom transformer aimed at recoding “marital status” in our dataset (basically we’ll do the same task we have already seen, implementing it with a custom transformer; for additional details on implementing customer estimator and transformer see the<span> </span><a rel="nofollow" href="https://www.oreilly.com/learning/extend-spark-ml-for-your-own-modeltransformer-types" target="_blank">nice article by H.Karau</a>. To define a custom transformer, we’ll define a new scala class,<span> </span><em>columnRecoder</em>, which extends the Transformer class, we’ll override the<span> </span><em>transformSchema</em>method to map the correct type of the new column we’re going to add with the transformation and we’ll implement the<span> </span><em>transform</em><span> </span>method which actually does the recoding in our dataset. A possible implementation is :</p>
<div>
<pre><code>
<span>import</span> org.<span>apache</span>.<span>spark</span>.<span>ml</span>.<span>Transformer</span>
<span>class</span> <span>columnRecoder</span>(<span>override</span> <span>val</span> uid: String) <span>extends</span> Transformer {
  <span>final</span> <span>val</span> inputCol= <span>new</span> Param[String](<span>this</span>, <span>&quot;inputCol&quot;</span>, <span>&quot;input column&quot;</span>)
  <span>final</span> <span>val</span> outputCol = <span>new</span> Param[String](<span>this</span>, <span>&quot;outputCol&quot;</span>, <span>&quot;output column&quot;</span>)

<span>def</span> <span>setInputCol</span>(value: String): <span>this</span>.<span>type</span> = <span>set</span>(inputCol, value)

<span>def</span> <span>setOutputCol</span>(value: String): <span>this</span>.<span>type</span> = <span>set</span>(outputCol, value)

<span>def</span> <span>this</span>() = <span>this</span>(Identifiable.<span>randomUID</span>(<span>&quot;columnRecoder&quot;</span>))

<span>def</span> <span>copy</span>(existingParam: ParamMap): columnRecoder = {<span>defaultCopy</span>(existingParam)}
<span>override</span> <span>def</span> <span>transformSchema</span>(schema: StructType): StructType = {
    <span>// Check inputCol type</span>
    <span>val</span> idx = schema.<span>fieldIndex</span>($(inputCol))
    <span>val</span> field = schema.<span>fields</span>(idx)
    <span>if</span> (field.<span>dataType</span> != StringType) {
      <span>throw</span> <span>new</span> Exception(s<span>&quot;Input type ${field.dataType} type mismatch: String expected&quot;</span>)
    }
    <span>// The return field</span>
    schema.<span>add</span>(<span>StructField</span>($(outputCol),StringType, <span>false</span>))
  }

<span>val</span> newcol_recode = <span>new</span> <span>marital_code</span>()

<span>private</span> <span>def</span> <span>newcol_recode</span>(str1: String): String = { 
    <span>var</span> nw_str = <span>&quot;noVal&quot;</span>
     <span>if</span> ((str1 == <span>&quot;Married-spouse-absent&quot;</span>) | (str1 ==<span>&quot;Married-AF-spouse&quot;</span>) 
        | (str1 == <span>&quot;Married-civ-spouse&quot;</span>)) 
       {nw_str = <span>&quot;Married&quot;</span> } 
        <span>else</span> <span>if</span> ((str1 == <span>&quot;Divorced&quot;</span>) | (str1 == <span>&quot;Never-married&quot;</span> ) 
        | (str1 == <span>&quot;Separated&quot;</span> ) | (str1 == <span>&quot;Widowed&quot;</span>)) 
          {nw_str = <span>&quot;Nonmarried&quot;</span> } 
        <span>else</span> { nw_str = str1}
    nw_str
  }
  
<span>private</span> <span>def</span> udfnewcol =  <span>udf</span>(newcol_recode.<span>recode</span>(_))
  
<span>def</span> <span>transform</span>(df: Dataset[_]): DataFrame = { 
  df.<span>withColumn</span>($(outputCol), <span>udfnewcol</span>(<span>df</span>($(inputCol)))) 
  }
}</code></pre>
</div>
<p>Once defined as a transformer, we can use it in our pipeline as the first stage.</p>
<div>
<pre><code>
<span>// define stages</span>
<span>val</span> new_marital = <span>new</span> <span>columnRecoder</span>().<span>setInputCol</span>(<span>&quot;marital&quot;</span>)
.<span>setOutputCol</span>(<span>&quot;new_marital&quot;</span>)

<span>val</span> new_workclassIdx = <span>new</span> <span>StringIndexer</span>().<span>setInputCol</span>(<span>&quot;new_workclass&quot;</span>)
.<span>setOutputCol</span>(<span>&quot;new_workclassIdx&quot;</span>).<span>setHandleInvalid</span>(<span>&quot;skip&quot;</span>)

<span>val</span> genderIdx = <span>new</span> <span>StringIndexer</span>().<span>setInputCol</span>(<span>&quot;gender&quot;</span>)
.<span>setOutputCol</span>(<span>&quot;genderIdx&quot;</span>).<span>setHandleInvalid</span>(<span>&quot;skip&quot;</span>)

<span>val</span> maritalIdx = <span>new</span> <span>StringIndexer</span>().<span>setInputCol</span>(<span>&quot;new_marital&quot;</span>)
.<span>setOutputCol</span>(<span>&quot;maritalIdx&quot;</span>).<span>setHandleInvalid</span>(<span>&quot;skip&quot;</span>)

.......</code></pre>
</div>
<p>A second step in building our pipeline is to assemble categorical indexes in a single vector, therefore many categorical indexes are put all together in a single vector through the<span> </span><em>VectorAssembler</em>transformer. This<span> </span><em>VectorAssembler</em><span> </span>will deliver a single column feature which will be, in turn, transformed to indexes by<span> </span><em>VectorIndexer</em><span> </span>transformer to deliver indexes within the<span> </span><em>“catFeatures”</em> column:</p>
<div>
<pre><code><span>// cat vector for categorical variables</span>
<span>val</span> catVect = <span>new</span> <span>VectorAssembler</span>()
                  .<span>setInputCols</span>(Array(<span>&quot;new_workclassIdx&quot;</span>, <span>&quot;genderIdx&quot;</span>, <span>&quot;catVect&quot;</span>,<span>&quot;maritalIdx&quot;</span>, <span>&quot;occupationIdx&quot;</span>,<span>&quot;relationshipIdx&quot;</span>,<span>&quot;raceIdx&quot;</span>,<span>&quot;countryIdx&quot;</span>))
                  .<span>setOutputCol</span>(<span>&quot;cat01Features&quot;</span>)

<span>val</span> catIdx = <span>new</span> <span>VectorIndexer</span>()
                  .<span>setInputCol</span>(catVect.<span>getOutputCol</span>)
                  .<span>setOutputCol</span>(<span>&quot;catFeatures&quot;</span>)</code></pre>
</div>
<p>For numeric variables we need just to assemble columns with VectorAssembler, then we’re ready to put these two vectors (one for categorical variables, the other for numeric variables) together in a single vector.</p>
<div>
<pre><code>
<span>// numeric vector for numeric variable</span>
<span>val</span> numVect = <span>new</span> <span>VectorAssembler</span>()
                  .<span>setInputCols</span>(Array(<span>&quot;age&quot;</span>,<span>&quot;hr_per_week&quot;</span>,<span>&quot;capitalgain&quot;</span>,<span>&quot;capitalloss&quot;</span>,<span>&quot;eduyears&quot;</span>))
                  .<span>setOutputCol</span>(<span>&quot;numFeatures&quot;</span>)

<span>val</span> featVect = <span>new</span> <span>VectorAssembler</span>()
                    .<span>setInputCols</span>(Array(<span>&quot;catFeatures&quot;</span>, <span>&quot;numFeatures&quot;</span>))
                    .<span>setOutputCol</span>(<span>&quot;features&quot;</span>)</code></pre>
</div>
<p>We have now label and features ready to build the logistic regression model which is the final component of our pipeline. We can also set some parameters for the model, in particular, we can define the threshold (by default set to 0.5) to make the decision between label values, as well as the max number of iterations for this algorithm and a parameter to tune regularization.<br>
When all stages of the pipeline are ready, we just need to define the<span> </span><em>pipeline</em><span> </span>component itself, passing as an input parameter an array with all defined stages:<span></span></p>
<div>
<pre><code><span>val</span> lr = <span>new</span> <span>LogisticRegression</span>().<span>setLabelCol</span>(<span>&quot;labelIdx&quot;</span>).<span>setFeaturesCol</span>(<span>&quot;features&quot;</span>)
  .<span>setThreshold</span>(<span>0.33</span>).<span>setMaxIter</span>(<span>10</span>).<span>setRegParam</span>(<span>0.2</span>)

<span>val</span> pipeline = <span>new</span> <span>Pipeline</span>().<span>setStages</span>(Array(new_marital,new_workclassIdx, labelIdx,maritalIdx,occupationIdx, relationshipIdx,raceIdx,genderIdx, countryIdx,catVect, catIdx, numVect,featVect,lr))</code></pre>
</div>
<p>Now the pipeline component, which encompasses a number of transformations as well as the classification algorithm, is ready; to actually use it we supply a<span> </span><em>train</em><span> </span>dataset to fit the model and then a<span> </span><em>test</em><span> </span>dataset to evaluate our fitted model. Since we have defined a pipeline, we’ll be sure that both, train and test datasets, will undergo the same transformations, therefore, we don’t have to replicate the process twice.<br>
We need now to define train and test datasets.In our dataset, label values are unbalanced being the “more than 50k USD per year” value around the 25% of the total, in order to preserve the same proportion between label values we’ll subset the original dataset based on label value, obtaining a<span> </span><em>low-income</em><span> </span>dataset and an<span> </span><em>high-income</em><span> </span>dataset. We’ll split both dataset for train (70%) and test (30%), then we’ll merge back the two “train”” and the two “test” datasets and we’ll use resulting “train” dataset as input for our pipeline:<span></span></p>
<div>
<pre><code><span>// split betwen train and test</span>
<span>val</span> df_low_income = df_work.<span>filter</span>(<span>&quot;label == '&lt;=50K'&quot;</span>)
<span>val</span> df_high_income = df_work.<span>filter</span>(<span>&quot;label == '>50K'&quot;</span>)

<span>val</span> splits_LI = df_low_income.<span>randomSplit</span>(Array(<span>0.7</span>, <span>0.3</span>), seed=<span>123</span>)
<span>val</span> splits_HI = df_high_income.<span>randomSplit</span>(Array(<span>0.7</span>, <span>0.3</span>), seed=<span>123</span>)

<span>val</span> df_work_train = <span>splits_LI</span>(<span>0</span>).<span>unionAll</span>(<span>splits_HI</span>(<span>0</span>))
<span>val</span> df_work_test = <span>splits_LI</span>(<span>1</span>).<span>unionAll</span>(<span>splits_HI</span>(<span>1</span>))

<span>// fitting the pipeline</span>
<span>val</span> data_model = pipeline.<span>fit</span>(df_work_train)</code></pre>
</div>
<p>Once the pipeline is trained, we can use the data_model for testing against the test dataset, calculate the confusion matrix and evaluate the classifier metrics :</p>
<div>
<pre><code><span>// generate prediction</span>
<span>val</span> data_prediction = data_model.<span>transform</span>(df_work_test)
<span>val</span> data_predicted = data_prediction.<span>select</span>(<span>&quot;features&quot;</span>, <span>&quot;prediction&quot;</span>, <span>&quot;label&quot;</span>,<span>&quot;labelIdx&quot;</span>)

<span>// confusion matrix</span>
<span>val</span> tp = data_predicted.<span>filter</span>(<span>&quot;prediction == 1 AND labelIdx == 1&quot;</span>).<span>count</span>().<span>toFloat</span>
<span>val</span> fp = data_predicted.<span>filter</span>(<span>&quot;prediction == 1 AND labelIdx == 0&quot;</span>).<span>count</span>().<span>toFloat</span>
<span>val</span> tn = data_predicted.<span>filter</span>(<span>&quot;prediction == 0 AND labelIdx == 0&quot;</span>).<span>count</span>().<span>toFloat</span>
<span>val</span> fn = data_predicted.<span>filter</span>(<span>&quot;prediction == 0 AND labelIdx == 1&quot;</span>).<span>count</span>().<span>toFloat</span>
<span>val</span> metrics = spark.<span>createDataFrame</span>(Seq(
 (<span>&quot;TP&quot;</span>, tp),
 (<span>&quot;FP&quot;</span>, fp),
 (<span>&quot;TN&quot;</span>, tn),
 (<span>&quot;FN&quot;</span>, fn),
 (<span>&quot;Precision&quot;</span>, tp / (tp + fp)),
 (<span>&quot;Recall&quot;</span>, tp / (tp + fn)))).<span>toDF</span>(<span>&quot;metric&quot;</span>, <span>&quot;value&quot;</span>)
metrics.<span>show</span>()</code></pre>
</div>
</div>
<pre><code></code></pre>
<div>
<h3>R code and SparklyR</h3>
<p>Now we’ll try to replicate the same example we just saw in R, more precisely, working with the SparklyR package.<br>
We’ll use the developer version of SparklyR (as you possibly know, there’s an interesting debate on the best API to access Apache Spark resources from R. For those that wants to know more about<span> </span><a rel="nofollow" href="https://github.com/rstudio/sparklyr/issues/502" target="_blank">https://github.com/rstudio/sparklyr/issues/502</a>,<span> </span><a rel="nofollow" href="http://blog.revolutionanalytics.com/2016/10/tutorial-scalable-r-on-spark.html" target="_blank">http://blog.revolutionanalytics.com/2016/10/tutorial-scalable-r-on-spark.html</a>).<br>
We need to install it from github before connecting with Spark environment.<br>
In our case Spark is a standalone instance running version 2.2.0, as reported in the official documentation for SparklyR, configuration parameters can be set through spark_config() function, in particular, spark_config() provides the basic configuration used by default for spark connection. To change parameters it’s possible to get default configuration via spark_connection() then change parameters as needed (<span> </span><a rel="nofollow" href="https://spark.rstudio.com/guides/connections/" target="_blank">here’s the link for additional details to run sparklyR on Yarn cluster</a>).</p>
<div>
<pre><code>devtools<span>::</span><span>install_github</span>(<span>&quot;rstudio/sparklyr&quot;</span>)
<span>library</span>(sparklyr)
<span>library</span>(dplyr)

sc &lt;-<span> </span><span>spark_connect</span>(<span>master =</span> <span>&quot;local&quot;</span>,<span>spark_home =</span> <span>&quot;...\Local\spark&quot;</span>,<span>version=</span><span>&quot;2.2.0&quot;</span>)</code></pre>
</div>
<p>After connecting with Spark, we’ll read the dataset into a Spark dataframe and select fields (with column renaming, where needed) we’re going to use for our classification example. It is worthwhile to remember that<span> </span><em>dplyr</em><span> </span>(and therefore sparklyR) uses lazy evaluation when accessing and manipulating data, which means that<span> </span><em>‘the data is only fetched at the last moment when it’s needed’</em><span> </span>(Efficient R programming,<span> </span><em>C.Gillespie R.Lovelace</em>).For this reason, later on we’ll that we’ll force the statement execution calling<span> </span><em>action</em><span> </span>functions (like<span> </span><em>collect()</em>).<br>
As we did in scala script, we’ll recode<span> </span><em>“marital status”</em><span> </span>and<span> </span><em>“workclass”</em><span> </span>in order to simplify the analysis. In renaming dataframe columns, we’ll use the “select” function available from dplyr package, indeed one of the SparklyR aims is to allow the manipulation of Spark dataframes/tables through dplyr functions. This is an example of how function like “select” or “filter” can be used also for spark dataframes.</p>
<div>
<pre><code>income_table &lt;-<span> </span><span>spark_read_csv</span>(sc,<span>&quot;income_table&quot;</span>,<span>&quot;...</span><span>\a</span><span>dultincome</span><span>\a</span><span>dult.csv&quot;</span>)

income_table &lt;-<span> </span><span>select</span>(income_table,<span>&quot;workclass&quot;</span>,<span>&quot;gender&quot;</span>,<span>&quot;eduyears&quot;</span>=<span>&quot;educationalnum&quot;</span>,
        <span>&quot;age&quot;</span>,<span>&quot;marital&quot;</span>=<span>&quot;maritalstatus&quot;</span>,<span>&quot;occupation&quot;</span>,<span>&quot;relationship&quot;</span>,
        <span>&quot;race&quot;</span>,<span>&quot;hr_per_week&quot;</span>=<span>&quot;hoursperweek&quot;</span>,<span>&quot;country&quot;</span>=<span>&quot;nativecountry&quot;</span>,
        <span>&quot;label&quot;</span>=<span>&quot;income&quot;</span>,<span>&quot;capitalgain&quot;</span>,<span>&quot;capitalloss&quot;</span>)

<span># recoding marital status and workingclass</span>

income_table &lt;-<span> </span>income_table <span>%>%</span><span> </span><span>mutate</span>(<span>marital =</span> <span>ifelse</span>(marital <span>==</span><span> &quot;Married-spouse-absent&quot;</span> <span>|</span><span> </span>marital <span>==</span><span> &quot;Married-AF-spouse&quot;</span> <span>|</span><span> </span>
marital <span>==</span><span> &quot;Married-civ-spouse&quot;</span>,<span>&quot;married&quot;</span>,<span>&quot;nonMarried&quot;</span>))

income_table &lt;-<span> </span>income_table <span>%>%</span><span> </span><span>mutate</span>(<span>workclass =</span> <span>ifelse</span>(workclass <span>==</span><span> &quot;Local-gov&quot;</span><span>|</span><span> </span>workclass <span>==</span><span> &quot;Federal-gov&quot;</span> <span>|</span><span> </span>workclass <span>==</span><span> &quot;State_gov&quot;</span>,
<span>&quot;Gov&quot;</span>,<span>ifelse</span>(workclass <span>==</span><span> &quot;Self-emp-inc&quot;</span> <span>|</span><span> </span>workclass <span>==</span><span> &quot;Self-emp-not-inc&quot;</span>,<span>&quot;Selfemployed&quot;</span>,<span>ifelse</span>(workclass<span>==</span><span>&quot;Never-worked&quot;</span> <span>|</span><span> </span>
workclass <span>==</span><span> &quot;Without-pay&quot;</span>,<span>&quot;Notemployed&quot;</span>,workclass))))   </code></pre>
</div>
<p>SparklyR provides functions which are bound to Spark spark.ml package, therefore it is possible to build Machine Learning solutions putting together the power of dplyr grammar with Spark ML algorithms. To simply link package function to Spark.ml, SparklyR provides functions which use specific prefixes to identify functions group:</p>
<ul>
<li>functions prefixed with<span> </span><strong>sdf_</strong><span> </span>generally access the Scala Spark DataFrame API directly (as opposed to the dplyr interface which uses Spark SQL) to manipulate dataframes;</li>
<li>functions prefixed with<span> </span><strong>ft_</strong><span> </span>are functions to manipulate and transform features. Pipeline transformers and estimators belong to this group of functions;</li>
<li>functions prefixed with<span> </span><strong>ml_</strong><span> </span>implement algorithms to build machine learning workflow. Even pipeline instance is provided by<span> </span><em>ml_pipeline()</em><span> </span>which belongs to these functions.</li>
</ul>
<p>We can then proceed with pipeline, stages and feature array definition. Ft-prefixed functions recall the spark.ml functions these are bound to:</p>
<div>
<pre><code>income_pipe &lt;-<span> </span><span>ml_pipeline</span>(sc,<span>uid=</span><span>&quot;income_pipe&quot;</span>)
income_pipe &lt;-<span>f</span><span>t_string_indexer</span>(income_pipe,<span>input_col=</span><span>&quot;workclass&quot;</span>,<span>output_col=</span><span>&quot;workclassIdx&quot;</span>)
income_pipe &lt;-<span> </span><span>ft_string_indexer</span>(income_pipe,<span>input_col=</span><span>&quot;gender&quot;</span>,<span>output_col=</span><span>&quot;genderIdx&quot;</span>)
income_pipe &lt;-<span> </span><span>ft_string_indexer</span>(income_pipe,<span>input_col=</span><span>&quot;marital&quot;</span>,<span>output_col=</span><span>&quot;maritalIdx&quot;</span>)
income_pipe &lt;-<span> </span><span>ft_string_indexer</span>(income_pipe,<span>input_col=</span><span>&quot;occupation&quot;</span>,<span>output_col=</span><span>&quot;occupationIdx&quot;</span>)
income_pipe &lt;-<span> </span><span>ft_string_indexer</span>(income_pipe,<span>input_col=</span><span>&quot;race&quot;</span>,<span>output_col=</span><span>&quot;raceIdx&quot;</span>)
income_pipe &lt;-<span> </span><span>ft_string_indexer</span>(income_pipe,<span>input_col=</span><span>&quot;country&quot;</span>,<span>output_col=</span><span>&quot;countryIdx&quot;</span>)

income_pipe &lt;-<span> </span><span>ft_string_indexer</span>(income_pipe,<span>input_col=</span><span>&quot;label&quot;</span>,<span>output_col=</span><span>&quot;labelIdx&quot;</span>)

array_features &lt;-<span> </span><span>c</span>(<span>&quot;workclassIdx&quot;</span>,<span>&quot;genderIdx&quot;</span>,<span>&quot;maritalIdx&quot;</span>,
                    <span>&quot;occupationIdx&quot;</span>,<span>&quot;raceIdx&quot;</span>,<span>&quot;countryIdx&quot;</span>,<span>&quot;eduyears&quot;</span>,
                    <span>&quot;age&quot;</span>,<span>&quot;hr_per_week&quot;</span>,<span>&quot;capitalgain&quot;</span>,<span>&quot;capitalloss&quot;</span>)

income_pipe &lt;-<span> </span><span>ft_vector_assembler</span>(income_pipe, <span>input_col=</span>array_features, <span>output_col=</span><span>&quot;features&quot;</span>)</code></pre>
</div>
<p>In our example we’ll use<span> </span><em>ml_logistic_regression()</em><span> </span>to implement the classification solution aimed at predicting the income level. Being bound to the<span> </span><em>LogisticRegression()</em><span> </span>function in spark.ml package, (as expected) all parameters are available for specific setting, therefore we can can “mirror” the same call we made in scala code: e.g. we can manage the “decision” threshold for our binary classifier (setting the value to 0.33).</p>
<div>
<pre><code><span># putting in pipe the logistic regression evaluator</span>
income_pipe &lt;-<span> </span><span>ml_logistic_regression</span>(income_pipe, 
<span>features_col =</span> <span>&quot;features&quot;</span>, <span>label_col =</span> <span>&quot;labelIdx&quot;</span>,
<span>family=</span> <span>&quot;binomial&quot;</span>,<span>threshold =</span> <span>0.33</span>, <span>reg_param=</span><span>0.2</span>, <span>max_iter=</span>10L)</code></pre>
</div>
<p>Final steps in our example are to split data between train and test subset, fit the pipeline and evaluate the classifier. As we’ve had already done in scala code, we’ll manage the unbalance in label values, splitting the dataframe in a way which secures the relative percentage of label values. Afterwards, we’ll fit the pipeline and get the predictions relative to test dataframe (as we did already in scala code).</p>
<div>
<pre><code><span># data split</span>
<span># dealing with label inbalance</span>

df_low_income =<span> </span><span>filter</span>(income_table,income_table<span>$</span>label <span>==</span><span> &quot;&lt;=50K&quot;</span>)
df_high_income =<span> </span><span>filter</span>(income_table,income_table<span>$</span>label <span>==</span><span> &quot;>50K&quot;</span>)

splits_LI &lt;-<span> </span><span>sdf_partition</span>(df_low_income, <span>test=</span><span>0.3</span>, <span>train=</span><span>0.7</span>, <span>seed =</span> <span>7711</span>)
splits_HI &lt;-<span> </span><span>sdf_partition</span>(df_high_income,<span>test=</span><span>0.3</span>,<span>train=</span><span>0.7</span>,<span>seed=</span><span>7711</span>)

df_test &lt;-<span> </span><span>sdf_bind_rows</span>(splits_LI[[<span>1</span>]],splits_HI[[<span>1</span>]])
df_train &lt;-<span> </span><span>sdf_bind_rows</span>(splits_LI[[<span>2</span>]],splits_HI[[<span>2</span>]])

df_model &lt;-<span> </span><span>ml_fit</span>(income_pipe,df_train)</code></pre>
</div>
<p>Once fitted, the model exposes, for each pipeline stage, all parameters and logistic regression (which is the last element in the stages list) coefficients.</p>
<div>
<pre><code>df_model<span>$</span>stages
df_model<span>$</span>stages[[<span>9</span>]]<span>$</span>coefficients</code></pre>
</div>
<p>We can then process the test dataset putting it in the pipeline to get predictions and evaluate the fitted model</p>
<div>
<pre><code>df_prediction &lt;-<span> </span><span>ml_predict</span>(df_model,df_test)
df_results &lt;-<span> </span><span>select</span>(df_prediction,<span>&quot;prediction&quot;</span>,<span>&quot;labelIdx&quot;</span>,<span>&quot;probability&quot;</span>)</code></pre>
</div>
<p>We can then proceed to evaluate the confusion matrix and get main metrics to evaluate the model (from precision to AUC).</p>
<div>
<pre><code><span># calculating confusion matrix</span>

df_tp &lt;-<span> </span><span>filter</span>(df_results,(prediction <span>==</span><span> </span><span>1</span> <span>&amp;&amp;</span><span> </span>labelIdx <span>==</span><span> </span><span>1</span>))
df_fp &lt;-<span> </span><span>filter</span>(df_results,(prediction <span>==</span><span>1</span> <span>&amp;&amp;</span><span> </span>labelIdx <span>==</span><span> </span><span>0</span>))
df_tn &lt;-<span> </span><span>filter</span>(df_results,(prediction <span>==</span><span> </span><span>1</span> <span>&amp;&amp;</span><span> </span>labelIdx <span>==</span><span> </span><span>0</span>))
df_fn &lt;-<span> </span><span>filter</span>(df_results,(prediction <span>==</span><span> </span><span>1</span> <span>&amp;&amp;</span><span> </span>labelIdx <span>==</span><span> </span><span>1</span>))


tp &lt;-<span> </span>df_tp <span>%>%</span><span> </span><span>tally</span>() <span>%>%</span><span> </span><span>collect</span>() <span>%>%</span><span> </span><span>as.integer</span>()
fp &lt;-<span> </span>df_fp <span>%>%</span><span> </span><span>tally</span>() <span>%>%</span><span> </span><span>collect</span>() <span>%>%</span><span> </span><span>as.integer</span>()
tn &lt;-<span> </span>df_tn <span>%>%</span><span> </span><span>tally</span>() <span>%>%</span><span> </span><span>collect</span>() <span>%>%</span><span> </span><span>as.integer</span>()
fn &lt;-<span> </span>df_fn <span>%>%</span><span> </span><span>tally</span>() <span>%>%</span><span> </span><span>collect</span>() <span>%>%</span><span> </span><span>as.integer</span>()

df_precision &lt;-<span> </span>(tp<span>/</span>(tp<span>+</span>fp))
df_recall &lt;-<span> </span>(tp<span>/</span>(tp<span>+</span>fn))
df_accuracy =<span> </span>(tp<span>+</span>tn)<span>/</span>(tp<span>+</span>tn<span>+</span>fp<span>+</span>fn)

c_AUC &lt;-<span> </span><span>ml_binary_classification_eval</span>(df_prediction, <span>label_col =</span> <span>&quot;labelIdx&quot;</span>,
<span>prediction_col =</span> <span>&quot;prediction&quot;</span>, <span>metric_name =</span> <span>&quot;areaUnderROC&quot;</span>)</code></pre>
</div>
</div>
<pre><code></code></pre>
<div>
<h3>Conclusion</h3>
<p>As we have seen, pipelines are a useful mechanism to assemble and serialize transformation in order to make it repeatable for different sets of data. Are then a simple way for fitting and evaluating models through train/test datasets, but also suitable to run the same sequence of transformer/estimator in parallel over different nodes of the Spark cluster (i.e. to find the best parameter set). Moreover, a powerful API to deal with pipelines in R is available by SparklyR package, which provides in addition a comprehensive set of functions to leverage the ML spark package (<a rel="nofollow" href="https://spark.rstudio.com/examples/" target="_blank">here’s a link for a guide to deploy SparlyR in different cluster environment</a>). Finally a support to run R code distributed across a Spark cluster has been to SparklyR with the<span> </span><em>spark_apply()</em><span> </span>function (<a rel="nofollow" href="https://spark.rstudio.com/guides/distributed-r/" target="_blank">https://spark.rstudio.com/guides/distributed-r/</a>) which makes evenmore interesting the possibility to leverage pipelines in R in ditributed environment for analytical solutions.
</div>
</div>
	<p>
<div>
<div>To <strong>leave a comment</strong> for the author, please follow the link and comment on their blog: <strong><a href="http://r-posts.com/leveraging-pipeline-in-spark-trough-scala-and-sparklyr/"> R-posts.com</a></strong>.</div>
<hr>
<a rel="nofollow" href="https://www.r-bloggers.com/">R-bloggers.com</a> offers <strong><a rel="nofollow" href="https://feedburner.google.com/fb/a/mailverify?uri=RBloggers">daily e-mail updates</a></strong> about <a rel="nofollow" href="https://www.r-project.org/" title="The R Project for Statistical Computing">R</a> news and <a rel="nofollow" href="https://www.r-bloggers.com/search/tutorial" title="R tutorials">tutorials</a> on topics such as: <a rel="nofollow" href="https://www.r-bloggers.com/search/data%20science" title="Data science">Data science</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/Big%20Data" title="Big Data">Big Data, <a rel="nofollow" href="https://www.r-users.com/" title="R jobs">R jobs</a>, visualization (<a rel="nofollow" href="https://www.r-bloggers.com/search/ggplot2" title="ggplot and ggplot2 tutorials">ggplot2</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/boxplot" title="Boxplots using lattice and ggplot2 tutorials">Boxplots</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/map" title="Maps and gis">maps</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/animation" title="Animation in R">animation</a>), programming (<a rel="nofollow" href="https://www.r-bloggers.com/search/RStudio" title="RStudio IDE for R">RStudio</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/sweave" title="Sweave and literate programming">Sweave</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/LaTeX" title="LaTeX in R">LaTeX</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/SQL" title="SQL and databases">SQL</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/eclipse" title="Eclipse IDE for R">Eclipse</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/git" title="git and github, Version Control System">git</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/hadoop" title="Large data in R using Hadoop">hadoop</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/Web+Scraping" title="Web Scraping of google, facebook, yahoo, twitter and more using R">Web Scraping</a>) statistics (<a rel="nofollow" href="https://www.r-bloggers.com/search/regression" title="Regressions and ANOVA analysis tutorials">regression</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/PCA" title="principal component analysis tutorial">PCA</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/time+series" title="Time series">time series</a>, <a rel="nofollow" href="https://www.r-bloggers.com/search/trading" title="finance trading">trading</a>) and more...
<img width="1" alt="" src="http://feeds.feedburner.com/~r/RBloggers/~4/vAJQqapRFYU" height="1"></a></div>