title: scala/spark map returning an array instead of single value
body: <p>I have a Scala class that is intended to generalize some functionality of linear models--specifically the user should be able to create an instance with an array of coefficients and an array of predictors, and the class pulls the data from a DataFrame, and uses a simple linear model to create predictions on an entire DataFrame as shown below.</p>
  
  <p>I am stuck on the last line... which I expect to generate a column of predicted values.  I have attempted a number of approaches (all but one of which are commented out). The code as it is now wont compile b/c of a type mismatch:</p>
  
  <pre><code>[error]  found   : Array[org.apache.spark.sql.Column]
  [error]  required: org.apache.spark.sql.Column
  [error]       .withColumn("prediction", colMod(preds.map(p =&gt; data(p))))
  [error]                                               ^
  </code></pre>
  
  <p>...which I also get in the pred &lt;- preds version... and the foreach version:</p>
  
  <pre><code>[error]  found   : Unit
  [error]  required: org.apache.spark.sql.Column
  [error]       .withColumn("prediction", colMod(preds.foreach(data(_))))
  [error]                                                   ^
  </code></pre>
  
  <p>Been trying in vain to resolve... would be grateful for any suggestions.</p>
  
  <pre><code>  class LinearModel(coefficients: Array[Double],
                      predictors: Array[String],
                      data: DataFrame) {
  
      val coefs = coefficients
      val preds = Array.concat(Array("bias"), predictors)
      require(coefs.length == preds.length)
  
      /**
        * predict: computes linear model predictions as the dot product of the coefficents and the
        * values (X[i] in the model matrix)
        * @param values: the values from a single row of the given variables from model matrix X
        * @param coefs: array of coefficients to be applied to each of the variables in values
        *             (the first coef is assumed to be 1 for the bias/intercept term)
        * @return: the predicted value 
        */
      private def predict(values: Array[Double], coefs: Array[Double]): Unit = {
        (for ((c, v) &lt;- coefs.zip(values)) yield c * v).sum
      }
  
      /**
        * colMod (udf): passes the values for each relevant value to predict()
        * @param values: an Array of the numerical values of each of the specified predictors for a
        *              given record
        */
      private val colMod = udf((values: Array[Double]) =&gt; predict(values, coefs))
  
      val dfPred = data
        // create the column with the prediction
        .withColumn("prediction", colMod(preds.map(p =&gt; data(p))))
        //.withColumn("prediction", colMod(for (pred &lt;- preds) yield data(pred)))
        //.withColumn("prediction", colMod(preds.foreach(data(_))))
        // prev line should = colMod(data(pred1), data(pred2), ..., data(predn))
    }
  </code></pre>