Spark - Convertir RDD [Vector] en DataFrame avec des colonnes variables

Question

Quelle est la meilleure solution pour généraliser la conversion de RDD [Vector] en DataFrame avec scala/spark 1.6. Les entrées sont différentes RDD [Vector]. Le nombre de colonnes dans Vector peut être compris entre 1 et n pour différents RDD.

J'ai essayé d'utiliser une bibliothèque sans forme, en leur indiquant le nombre et le type des colonnes déclarées. ES:

val df = rddVector.map(_.toArray.toList) 
    .collect { 
      case t: List[Double] if t.length == 3 => t.toHList[Double :: Double :: Double :: HNil].get.tupled.productArity 
    } 
    .toDF("column_1", "column_2", "column_3") 

Merci!

Answer 1

Cela a fonctionné pour moi.

// Create a vector rdd 
    val vectorRDD = sc.parallelize(Seq(Seq(123L, 345L), Seq(567L, 789L), Seq(567L, 789L, 233334L))). 
    map(s => Vectors.dense(s.toSeq.map(_.toString.toDouble).toArray)) 

    // Calculate the maximum length of the vector to create a schema 
    val vectorLength = vectorRDD.map(x => x.toArray.length).max() 

    // create the dynamic schema 
    var schema = new StructType() 
    var i = 0 
    while (i < vectorLength) { 
    schema = schema.add(StructField(s"val${i}", DoubleType, true)) 
    i = i + 1 
    } 

    // create a rowRDD variable and make each row have the same arity 
    val rowRDD = vectorRDD.map { x => 
    var row = new Array[Double](vectorLength) 
    val newRow = x.toArray 

    System.arraycopy(newRow, 0, row, 0, newRow.length); 

    println(row.length) 

    Row.fromSeq(row) 
    } 

    // create your dataframe 
    val dataFrame = sqlContext.createDataFrame(rowRDD, schema) 

Sortie:

root 
|-- val0: double (nullable = true) 
|-- val1: double (nullable = true) 
|-- val2: double (nullable = true) 

+-----+-----+--------+ 
| val0| val1| val2| 
+-----+-----+--------+ 
|123.0|345.0|  0.0| 
|567.0|789.0|  0.0| 
|567.0|789.0|233334.0| 
+-----+-----+--------+