Spark Scala: How to convert Dataframe[vector] to DataFrame[f1:Double, …, fn: Double)]

by

Spark >= 3.0.0

Since Spark 3.0 you can use vector_to_array

import org.apache.spark.ml.functions.vector_to_array

testDF.select(vector_to_array($"scaledFeatures").alias("_tmp")).select(exprs:_*)

Spark < 3.0.0

One possible approach is something similar to this

import org.apache.spark.sql.functions.udf

// In Spark 1.x you'll will have to replace ML Vector with MLLib one
// import org.apache.spark.mllib.linalg.Vector
// In 2.x the below is usually the right choice
import org.apache.spark.ml.linalg.Vector

// Get size of the vector
val n = testDF.first.getAs[Vector](0).size

// Simple helper to convert vector to array<double> 
// asNondeterministic is available in Spark 2.3 or befor
// It can be removed, but at the cost of decreased performance
val vecToSeq = udf((v: Vector) => v.toArray).asNondeterministic

// Prepare a list of columns to create
val exprs = (0 until n).map(i => $"_tmp".getItem(i).alias(s"f$i"))

testDF.select(vecToSeq($"scaledFeatures").alias("_tmp")).select(exprs:_*)

If you know a list of columns upfront you can simplify this a little:

val cols: Seq[String] = ???
val exprs = cols.zipWithIndex.map{ case (c, i) => $"_tmp".getItem(i).alias(c) }

For Python equivalent see How to split Vector into columns – using PySpark.

More Related Contents:

  1. Dropping a nested column from Spark DataFrame
  2. MatchError while accessing vector column in Spark 2.0
  3. How to define a custom aggregation function to sum a column of Vectors?
  4. How to query JSON data column using Spark DataFrames?
  5. Encoder error while trying to map dataframe row to updated row
  6. How do I detect if a Spark DataFrame has a column
  7. How to define schema for custom type in Spark SQL?
  8. How to avoid duplicate columns after join?
  9. Flattening Rows in Spark
  10. How can I change column types in Spark SQL’s DataFrame?
  11. Spark unionAll multiple dataframes
  12. How can I pass extra parameters to UDFs in Spark SQL?
  13. How to split a dataframe into dataframes with same column values?
  14. How to zip two (or more) DataFrame in Spark
  15. Spark DataFrame: does groupBy after orderBy maintain that order?
  16. Better way to convert a string field into timestamp in Spark
  17. Spark UDAF with ArrayType as bufferSchema performance issues
  18. Apache Spark how to append new column from list/array to Spark dataframe
  19. How to use COGROUP for large datasets
  20. Why does join fail with “java.util.concurrent.TimeoutException: Futures timed out after [300 seconds]”?
  21. how to filter out a null value from spark dataframe
  22. What are the various join types in Spark?
  23. Provide schema while reading csv file as a dataframe in Scala Spark
  24. Spark: Add column to dataframe conditionally
  25. Replace missing values with mean – Spark Dataframe
  26. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  27. How to calculate the size of dataframe in bytes in Spark?
  28. Spark-Monotonically increasing id not working as expected in dataframe?
  29. Calculate Cosine Similarity Spark Dataframe
  30. Difference between two rows in Spark dataframe

Leave a Comment