1

Following on from my earlier question, Convert a Spark Vector of features into an array, I've made progress: 

def extractUdf = udf((v: SDV) => v.toArray)
val temp: DataFrame = dataWithFeatures.withColumn("extracted_features", extractUdf($"features"))

temp.printSchema()

val featuresArray1: Array[Double] = temp.rdd.map(r => r.getAs[Double](0)).collect
val featuresArray2: Array[Double] = temp.rdd.map(r => r.getAs[Double](1)).collect
val featuresArray3: Array[Double] = temp.rdd.map(r => r.getAs[Double](2)).collect

val allfeatures: Array[Array[Double]] = Array(featuresArray1, featuresArray2, featuresArray3)
val flatfeatures: Array[Double] = allfeatures.flatten

This seems to give the result I want. The extractUdf function turns feature: Vector into extracted_feature: 

 |-- features: vector (nullable = true)
 |-- extracted_features: array (nullable = true)
|    |-- element: double (containsNull = false)

However, I don't understand why my next 3 lines of code (i.e. array featuresArray1, featuresArray2, featuresArray3) are picking up extracted_features as opposed to any other column in temp (like features) for example, and how to pick up the indices of the array (0,1,2) in a which directly references the number of features and is not hard-coded. Thanks for your help!

LucieCBurgess
  • 759
  • 5
  • 12
  • 26

1 Answers1

3

Lets say you have a dataframe 

+---+-------------+
|id |features     |
+---+-------------+
|1  |[1.0,2.0,3.0]|
|2  |[3.0,4.0,8.0]|
+---+-------------+

with schema

root
 |-- id: integer (nullable = false)
 |-- features: vector (nullable = true)

and you've extracted the vector feature to Array by doing 

import org.apache.spark.sql.functions._
import org.apache.spark.mllib.linalg.DenseVector
def extractUdf = udf((v: DenseVector) => v.toArray)
val temp = dataWithFeatures.withColumn("extracted_features", extractUdf($"features"))

which would give 

+---+-------------+------------------+
|id |features     |extracted_features|
+---+-------------+------------------+
|1  |[1.0,2.0,3.0]|[1.0, 2.0, 3.0]   |
|2  |[3.0,4.0,8.0]|[3.0, 4.0, 8.0]   |
+---+-------------+------------------+

root
 |-- id: integer (nullable = false)
 |-- features: vector (nullable = true)
 |-- extracted_features: array (nullable = true)
 |    |-- element: double (containsNull = false)

now referencing elements from extracted_features Array column is as with other array types in scala . So you can do 

temp.withColumn("firstValue", $"extracted_features"(0))
  .withColumn("secondValue", $"extracted_features"(1))
  .withColumn("thirdValue", $"extracted_features"(2))

which would give you 

+---+-------------+------------------+----------+-----------+----------+
|id |features     |extracted_features|firstValue|secondValue|thirdValue|
+---+-------------+------------------+----------+-----------+----------+
|1  |[1.0,2.0,3.0]|[1.0, 2.0, 3.0]   |1.0       |2.0        |3.0       |
|2  |[3.0,4.0,8.0]|[3.0, 4.0, 8.0]   |3.0       |4.0        |8.0       |
+---+-------------+------------------+----------+-----------+----------+

root
 |-- id: integer (nullable = false)
 |-- features: vector (nullable = true)
 |-- extracted_features: array (nullable = true)
 |    |-- element: double (containsNull = false)
 |-- firstValue: double (nullable = true)
 |-- secondValue: double (nullable = true)
 |-- thirdValue: double (nullable = true)

I hope the answer is helpful

Ramesh Maharjan
  • 41,071
  • 6
  • 69
  • 97
  • You are absolutely right, thank you :-) Very helpful! I have also realised that I was looking slightly in the wrong direction. The features vector used by predictRaw() in the Spark abstract ClassificationModel class, which must be implemented in a concrete class, is actually specifying a vector of features for a single sample, and gives an output which is a vector prediction for a single label. This is not at all clear from the documentation. I'm using Spark 2.2.0 in case anyone would like to avoid the same pitfall. – LucieCBurgess Sep 07 '17 at 09:42