Comment agréger les valeurs dans la collection après groupBy?

Spark >= 2.4

Vous pouvez remplacer flatten udf avec construit-dans aplatir l' function

import org.apache.spark.sql.functions.flatten

en laissant le reste comme elle est.

Spark >= 2.0, < 2.4

C'est possible, mais assez cher. En utilisant les données que vous avez fournies:

case class Record(
    visitorId: String, trackingIds: Array[String], emailIds: Array[String])

val df = Seq(
  Record("a158", Array("666b"), Array("12")),
  Record("7g21", Array("c0b5"), Array("45")),
  Record("7g21", Array("c0b4"), Array("87")),
  Record("a158", Array("666b",  "777c"), Array.empty[String])).toDF

et une fonction d'assistance:

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

val flatten = udf((xs: Seq[Seq[String]]) => xs.flatten)

nous pouvons remplir les espaces vides avec des espaces réservés:

import org.apache.spark.sql.functions.{array, lit, when}

val dfWithPlaceholders = df.withColumn(
  "emailIds", 
  when(size($"emailIds") === 0, array(lit(""))).otherwise($"emailIds"))

collect_lists et flatten:

import org.apache.spark.sql.functions.{array, collect_list}

val emailIds = flatten(collect_list($"emailIds")).alias("emailIds")
val trackingIds = flatten(collect_list($"trackingIds")).alias("trackingIds")

df
  .groupBy($"visitorId")
  .agg(trackingIds, emailIds)

//+---------+------------------+--------+
//|visitorId|       trackingIds|emailIds|
//+---------+------------------+--------+
//|     a158|[666b, 666b, 777c]|  [12, ]|
//|     7g21|      [c0b5, c0b4]|[45, 87]|
//+---------+------------------+--------+

Avec statiquement typé Dataset:

df.as[Record]
  .groupByKey(_.visitorId)
  .mapGroups { case (key, vs) => 
    vs.map(v => (v.trackingIds, v.emailIds)).toArray.unzip match {
      case (trackingIds, emailIds) => 
        Record(key, trackingIds.flatten, emailIds.flatten)
  }}

//+---------+------------------+--------+
//|visitorId|       trackingIds|emailIds|
//+---------+------------------+--------+
//|     a158|[666b, 666b, 777c]|  [12, ]|
//|     7g21|      [c0b5, c0b4]|[45, 87]|
//+---------+------------------+--------+

Étincelle 1.x

Vous pouvez convertir le RDD et le groupe

import org.apache.spark.sql.Row

dfWithPlaceholders.rdd
  .map {
     case Row(id: String, 
       trcks: Seq[String @ unchecked],
       emails: Seq[String @ unchecked]) => (id, (trcks, emails))
  }
  .groupByKey
  .map {case (key, vs) => vs.toArray.unzip match {
    case (trackingIds, emailIds) => 
      Record(key, trackingIds.flatten, emailIds.flatten)
  }}
  .toDF

//+---------+------------------+--------+
//|visitorId|       trackingIds|emailIds|
//+---------+------------------+--------+
//|     7g21|      [c0b5, c0b4]|[45, 87]|
//|     a158|[666b, 666b, 777c]|  [12, ]|
//+---------+------------------+--------+

Vous pouvez utiliser défini par l'Utilisateur agrégées fonctions.

1) créer un personnalisé udaf, nous à l'aide de la scala de classe appelé customAggregation.

package com.package.name
import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types._
import scala.collection.JavaConverters._
class CustomAggregation() extends UserDefinedAggregateFunction {
//Input Data Type Schema
def inputSchema: StructType = StructType(Array(StructField("col5", ArrayType(StringType))))
//Intermediate Schema
def bufferSchema = StructType(Array(
StructField("col5_collapsed",  ArrayType(StringType))))
//Returned Data Type .
def dataType: DataType = ArrayType(StringType)
//Self-explaining
def deterministic = true
//This function is called whenever key changes
def initialize(buffer: MutableAggregationBuffer) = {
buffer(0) = Array.empty[String] //initialize array
}
//Iterate over each entry of a group
def update(buffer: MutableAggregationBuffer, input: Row) = {
buffer(0) =
if(!input.isNullAt(0))
buffer.getList[String](0).toArray ++ input.getList[String](0).toArray
else
buffer.getList[String](0).toArray
}
//Merge two partial aggregates
def merge(buffer1: MutableAggregationBuffer, buffer2: Row) = {
buffer1(0) = buffer1.getList[String](0).toArray ++ buffer2.getList[String](0).toArray
}
//Called after all the entries are exhausted.
def evaluate(buffer: Row) = {
buffer.getList[String](0).asScala.toList.distinct
}
}

2) Puis utiliser l'udaf, nous, dans votre code comme

//define UDAF
val CustomAggregation = new CustomAggregation()
DataFrame
.groupBy(col1,col2,col3)
.agg(CustomAggregation(DataFrame(col5))).show()