перегруппировка/конкатенации DataFrame строки в Спарк

Question

У меня есть DataFrame, который выглядит следующим образом:

scala> data.show 
+-----+---+---------+ 
|label| id| features| 
+-----+---+---------+ 
| 1.0| 1|[1.0,2.0]| 
| 0.0| 2|[5.0,6.0]| 
| 1.0| 1|[3.0,4.0]| 
| 0.0| 2|[7.0,8.0]| 
+-----+---+---------+ 

Я хочу, чтобы перегруппировать функции на основе «ид», так что я могу получить следующее:

scala> data.show 
+---------+---+-----------------+ 
| label| id| features  | 
+---------+---+-----------------+ 
| 1.0,1.0| 1|[1.0,2.0,3.0,4.0]| 
| 0.0,0.0| 2|[5.0,6.0,7.8,8.0]| 
+---------+---+-----------------+ 

Это код, я использую для создания упомянутого DataFrame

val rdd = sc.parallelize(List((1.0, 1, Vectors.dense(1.0, 2.0)), (0.0, 2, Vectors.dense(5.0, 6.0)), (1.0, 1, Vectors.dense(3.0, 4.0)), (0.0, 2, Vectors.dense(7.0, 8.0)))) 
val data = rdd.toDF("label", "id", "features") 

Я пытался разные вещи с б od RDD и DataFrames. Самый «перспективный» подход до сих пор был в фильтре на основе «ID»

data.filter($"id".equalTo(1)) 

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

Но у меня есть два узких места сейчас:

1) Как автоматизировать фильтрацию для всех различных значений, что «идентификатор» мог бы иметь?

Ниже генерирует ошибку:

data.select("id").distinct.foreach(x => data.filter($"id".equalTo(x))) 

2) Как конкатенации общего "особенность" по отношению к данному "ID". Не пробовал много, так как я все еще застряли на 1)

Любое предложение более чем приветствуется

Примечание: Для осветления «ярлык» всегда одинакова для каждого вхождения «ID». Извините за путаницу, простое расширение моей задачи будет также группировать «метки» (обновленный пример)

Answer 1

Я считаю, что нет эффективного способа добиться того, чего вы хотите, и требования к дополнительным заказам не делают ситуация лучше. Чистым способом я могу думать о том, как это groupByKey:

import org.apache.spark.mllib.linalg.{Vectors, Vector} 
import org.apache.spark.sql.functions.monotonicallyIncreasingId 
import org.apache.spark.sql.Row 
import org.apache.spark.rdd.RDD 


val pairs: RDD[((Double, Int), (Long, Vector))] = data 
    // Add row identifiers so we can keep desired order 
    .withColumn("uid", monotonicallyIncreasingId) 
    // Create PairwiseRDD where (label, id) is a key 
    // and (row-id, vector is a value) 
    .map{case Row(label: Double, id: Int, v: Vector, uid: Long) => 
    ((label, id), (uid, v))} 

val rows = pairs.groupByKey.mapValues(xs => { 
    val vs = xs 
    .toArray 
    .sortBy(_._1) // Sort by row id to keep order 
    .flatMap(_._2.toDense.values) // flatmap vector values 

    Vectors.dense(vs) // return concatenated vectors 

}).map{case ((label, id), v) => (label, id, v)} // Reshape 

val grouped = rows.toDF("label", "id", "features") 

grouped.show 

// +-----+---+-----------------+ 
// |label| id|   features| 
// +-----+---+-----------------+ 
// | 0.0| 2|[5.0,6.0,7.0,8.0]| 
// | 1.0| 1|[1.0,2.0,3.0,4.0]| 
// +-----+---+-----------------+ 

Также можно использовать UDAF, подобную той, которую я уже предложенной для SPARK SQL replacement for mysql GROUP_CONCAT aggregate function но это еще менее эффективным, чем это.