How do I convert an RDD with a SparseVector Column to a DataFrame with a column as Vector
You have to use VectorUDT
here:
# In Spark 1.x
# from pyspark.mllib.linalg import SparseVector, VectorUDT
from pyspark.ml.linalg import SparseVector, VectorUDT
temp_rdd = sc.parallelize([
(0.0, SparseVector(4, {1: 1.0, 3: 5.5})),
(1.0, SparseVector(4, {0: -1.0, 2: 0.5}))])
schema = StructType([
StructField("label", DoubleType(), True),
StructField("features", VectorUDT(), True)
])
temp_rdd.toDF(schema).printSchema()
## root
## |-- label: double (nullable = true)
## |-- features: vector (nullable = true)
Just for completeness Scala equivalent:
import org.apache.spark.sql.Row
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{DoubleType, StructType}
// In Spark 1x.
// import org.apache.spark.mllib.linalg.{Vectors, VectorUDT}
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.linalg.SQLDataTypes.VectorType
val schema = new StructType()
.add("label", DoubleType)
// In Spark 1.x
//.add("features", new VectorUDT())
.add("features",VectorType)
val temp_rdd: RDD[Row] = sc.parallelize(Seq(
Row(0.0, Vectors.sparse(4, Seq((1, 1.0), (3, 5.5)))),
Row(1.0, Vectors.sparse(4, Seq((0, -1.0), (2, 0.5))))
))
spark.createDataFrame(temp_rdd, schema).printSchema
// root
// |-- label: double (nullable = true)
// |-- features: vector (nullable = true)
While @zero323 answer https://stackoverflow.com/a/32745924/1333621 makes sense, and I wish it worked for me - the rdd underlying the dataframe, sqlContext.createDataFrame(temp_rdd, schema), the still contained SparseVectors types I had to do the following to convert to DenseVector types - if someone has a shorter/better way I want to know
temp_rdd = sc.parallelize([
(0.0, SparseVector(4, {1: 1.0, 3: 5.5})),
(1.0, SparseVector(4, {0: -1.0, 2: 0.5}))])
schema = StructType([
StructField("label", DoubleType(), True),
StructField("features", VectorUDT(), True)
])
temp_rdd.toDF(schema).printSchema()
df_w_ftr = temp_rdd.toDF(schema)
print 'original convertion method: ',df_w_ftr.take(5)
print('\n')
temp_rdd_dense = temp_rdd.map(lambda x: Row(label=x[0],features=DenseVector(x[1].toArray())))
print type(temp_rdd_dense), type(temp_rdd)
print 'using map and toArray:', temp_rdd_dense.take(5)
temp_rdd_dense.toDF().show()
root
|-- label: double (nullable = true)
|-- features: vector (nullable = true)
original convertion method: [Row(label=0.0, features=SparseVector(4, {1: 1.0, 3: 5.5})), Row(label=1.0, features=SparseVector(4, {0: -1.0, 2: 0.5}))]
<class 'pyspark.rdd.PipelinedRDD'> <class 'pyspark.rdd.RDD'>
using map and toArray: [Row(features=DenseVector([0.0, 1.0, 0.0, 5.5]), label=0.0), Row(features=DenseVector([-1.0, 0.0, 0.5, 0.0]), label=1.0)]
+------------------+-----+
| features|label|
+------------------+-----+
| [0.0,1.0,0.0,5.5]| 0.0|
|[-1.0,0.0,0.5,0.0]| 1.0|
+------------------+-----+