Spark: Gruppo concat equivalente in scala RDD

Question

Ho seguito dataframe:

|-----id-------|----value------|-----desc------| 
    |  1  |  v1  |  d1  | 
    |  1  |  v2  |  d2  | 
    |  2  |  v21  |  d21  | 
    |  2  |  v22  |  d22  | 
    |--------------|---------------|---------------| 

voglio trasformarlo in:

|-----id-------|----value------|-----desc------| 
    |  1  |  v1;v2  |  d1;d2 | 
    |  2  |  v21;v22 |  d21;d22 | 
    |--------------|---------------|---------------| 

• E 'possibile attraverso operazioni di cornice di dati?
• Come sarebbe la trasformazione di rdd in questo caso?

Suppongo che rdd.reduce sia la chiave, ma non ho idea di come adattarlo a questo scenario.

Answer 1

È possibile trasformare i dati utilizzando scintilla SQL

case class Test(id: Int, value: String, desc: String) 
val data = sc.parallelize(Seq((1, "v1", "d1"), (1, "v2", "d2"), (2, "v21", "d21"), (2, "v22", "d22"))) 
    .map(line => Test(line._1, line._2, line._3)) 
    .df 

data.registerTempTable("data") 
val result = sqlContext.sql("select id,concat_ws(';', collect_list(value)),concat_ws(';', collect_list(value)) from data group by id") 
result.show 

Answer 2

Supponiamo di avere qualcosa di simile

import scala.util.Random 

val sqlc: SQLContext = ??? 

case class Record(id: Long, value: String, desc: String) 

val testData = for { 
    (i, j) <- List.fill(30)(Random.nextInt(5), Random.nextInt(5)) 
    } yield Record(i, s"v$i$j", s"d$i$j") 

val df = sqlc.createDataFrame(testData) 

Si può facilmente unire i dati come:

import sqlc.implicits._ 

def aggConcat(col: String) = df 
     .map(row => (row.getAs[Long]("id"), row.getAs[String](col))) 
     .aggregateByKey(Vector[String]())(_ :+ _, _ ++ _) 

val result = aggConcat("value").zip(aggConcat("desc")).map{ 
     case ((id, value), (_, desc)) => (id, value, desc) 
    }.toDF("id", "values", "descs") 

Se si desidera avere stringhe concatenate, invece di array, è possibile eseguire in seguito

import org.apache.spark.sql.functions._ 

val resultConcat = result 
     .withColumn("values", concat_ws(";", $"values")) 
     .withColumn("descs" , concat_ws(";", $"descs")) 

Answer 3

Dopo alcune ricerche mi è venuta in mente sth del genere:

val data = sc.parallelize(
    List(
     ("1", "v1", "d1"), 
     ("1", "v2", "d2"), 
     ("2", "v21", "d21"), 
     ("2", "v22", "d22"))) 
     .map{ case(id, value, desc)=>((id), (value, desc))} 
     .reduceByKey((x,y)=>(x._1+";"+y._1, x._2+";"+x._2)) 
     .map{ case(id,(value, desc))=>(id, value, desc)}.toDF("id", "value","desc") 
     .show() 

che mi lascia con:

+---+-------+-------+ 
    | id| value| desc| 
    +---+-------+-------+ 
    | 1| v1;v2| d1;d1| 
    | 2|v21;v22|d21;d21| 
    +---+-------+-------+ 

Answer 4

Se si lavora con DataFrames, utilizzare UDAF

import org.apache.spark.sql.Row 
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction} 
import org.apache.spark.sql.types.{DataType, StringType, StructField, StructType} 

class ConcatStringsUDAF(InputColumnName: String, sep:String = ",") extends UserDefinedAggregateFunction { 
    def inputSchema: StructType = StructType(StructField(InputColumnName, StringType) :: Nil) 
    def bufferSchema: StructType = StructType(StructField("concatString", StringType) :: Nil) 
    def dataType: DataType = StringType 
    def deterministic: Boolean = true 
    def initialize(buffer: MutableAggregationBuffer): Unit = buffer(0) = "" 

    private def concatStrings(str1: String, str2: String): String = { 
    (str1, str2) match { 
     case (s1: String, s2: String) => Seq(s1, s2).filter(_ != "").mkString(sep) 
     case (null, s: String) => s 
     case (s: String, null) => s 
     case _ => "" 
    } 
    } 
    def update(buffer: MutableAggregationBuffer, input: Row): Unit = { 
    val acc1 = buffer.getAs[String](0) 
    val acc2 = input.getAs[String](0) 
    buffer(0) = concatStrings(acc1, acc2) 
    } 

    def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = { 
    val acc1 = buffer1.getAs[String](0) 
    val acc2 = buffer2.getAs[String](0) 
    buffer1(0) = concatStrings(acc1, acc2) 
    } 

    def evaluate(buffer: Row): Any = buffer.getAs[String](0) 
} 

E poi usare in questo modo

val stringConcatener = new ConcatStringsUDAF("Category_ID", ",") 
data.groupBy("aaid", "os_country").agg(stringConcatener(data("X")).as("Xs")) 

A partire da Spark 1.6, dare un'occhiata a Dataset and Aggregator.