dataframe/Dataset groupBy comportamento/ottimizzazione

Question

Supponiamo di avere dataframe df composta dalle seguenti colonne:

Nome, Cognome, dimensioni, larghezza, lunghezza, pesare

Ora vogliamo effettuare una Un paio di operazioni, per esempio, vogliamo creare un paio di DataFrames contenenti dati su Dimensioni e Larghezza.

val df1 = df.groupBy("surname").agg(sum("size")) 
val df2 = df.groupBy("surname").agg(sum("width")) 

Come si può notare, altre colonne, come Lunghezza, non vengono utilizzate da nessuna parte. Spark è abbastanza intelligente da abbandonare le colonne ridondanti prima della fase di mescolamento o vengono trasportate in giro? Wil:

val dfBasic = df.select("surname", "size", "width") 

prima che il raggruppamento influenzi in qualche modo la prestazione?

Answer 1

Sì, è "sufficientemente intelligente". groupBy eseguito su un DataFrame non è la stessa operazione di groupBy eseguita su un semplice RDD. In uno scenario che hai descritto non è necessario spostare i dati grezzi. Creiamo un piccolo esempio per illustrare che:

val df = sc.parallelize(Seq(
    ("a", "foo", 1), ("a", "foo", 3), ("b", "bar", 5), ("b", "bar", 1) 
)).toDF("x", "y", "z") 

df.groupBy("x").agg(sum($"z")).explain 

// == Physical Plan == 
// *HashAggregate(keys=[x#148], functions=[sum(cast(z#150 as bigint))]) 
// +- Exchange hashpartitioning(x#148, 200) 
// +- *HashAggregate(keys=[x#148], functions=[partial_sum(cast(z#150 as bigint))]) 
//  +- *Project [_1#144 AS x#148, _3#146 AS z#150] 
//   +- *SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, scala.Tuple3, true])._1, true, false) AS _1#144, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, scala.Tuple3, true])._2, true, false) AS _2#145, assertnotnull(input[0, scala.Tuple3, true])._3 AS _3#146] 
//    +- Scan ExternalRDDScan[obj#143] 

Come potete la prima fase è una proiezione in cui si conservano solo le colonne necessarie. I dati successivi vengono aggregati localmente e infine trasferiti e aggregati a livello globale. Otterrai un output di risposta leggermente diverso se usi Spark < = 1.4 ma la struttura generale dovrebbe essere esattamente la stessa.

Infine una visualizzazione DAG mostrando che sopra descrizione descrive Lavoro attuale:

Analogamente, Dataset.groupByKey seguita da reduceGroups, contiene sia cartina facciata (ObjectHashAggregate con partial_reduceaggregator) e ridurre facciata (ObjectHashAggregate con reduceaggregator riduzione):

case class Foo(x: String, y: String, z: Int) 

val ds = df.as[Foo] 
ds.groupByKey(_.x).reduceGroups((x, y) => x.copy(z = x.z + y.z)).explain 

// == Physical Plan == 
// ObjectHashAggregate(keys=[value#126], functions=[reduceaggregator(org.apache.spark.sql.expressions.ReduceAggregator@54d90261, Some(newInstance(class $line40.$read$$iw$$iw$Foo)), Some(class $line40.$read$$iw$$iw$Foo), Some(StructType(StructField(x,StringType,true), StructField(y,StringType,true), StructField(z,IntegerType,false))), input[0, scala.Tuple2, true]._1 AS value#128, if ((isnull(input[0, scala.Tuple2, true]._2) || None.equals)) null else named_struct(x, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]._2)).x, true, false) AS x#25, y, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]._2)).y, true, false) AS y#26, z, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]._2)).z AS z#27) AS _2#129, newInstance(class scala.Tuple2), staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, $line40.$read$$iw$$iw$Foo, true])).x, true, false) AS x#25, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, $line40.$read$$iw$$iw$Foo, true])).y, true, false) AS y#26, assertnotnull(assertnotnull(input[0, $line40.$read$$iw$$iw$Foo, true])).z AS z#27, StructField(x,StringType,true), StructField(y,StringType,true), StructField(z,IntegerType,false), true, 0, 0)]) 
// +- Exchange hashpartitioning(value#126, 200) 
// +- ObjectHashAggregate(keys=[value#126], functions=[partial_reduceaggregator(org.apache.spark.sql.expressions.ReduceAggregator@54d90261, Some(newInstance(class $line40.$read$$iw$$iw$Foo)), Some(class $line40.$read$$iw$$iw$Foo), Some(StructType(StructField(x,StringType,true), StructField(y,StringType,true), StructField(z,IntegerType,false))), input[0, scala.Tuple2, true]._1 AS value#128, if ((isnull(input[0, scala.Tuple2, true]._2) || None.equals)) null else named_struct(x, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]._2)).x, true, false) AS x#25, y, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]._2)).y, true, false) AS y#26, z, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]._2)).z AS z#27) AS _2#129, newInstance(class scala.Tuple2), staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, $line40.$read$$iw$$iw$Foo, true])).x, true, false) AS x#25, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, $line40.$read$$iw$$iw$Foo, true])).y, true, false) AS y#26, assertnotnull(assertnotnull(input[0, $line40.$read$$iw$$iw$Foo, true])).z AS z#27, StructField(x,StringType,true), StructField(y,StringType,true), StructField(z,IntegerType,false), true, 0, 0)]) 
//  +- AppendColumns <function1>, newInstance(class $line40.$read$$iw$$iw$Foo), [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true, false) AS value#126] 
//   +- *Project [_1#4 AS x#8, _2#5 AS y#9, _3#6 AS z#10] 
//    +- *SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, scala.Tuple3, true])._1, true, false) AS _1#4, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, scala.Tuple3, true])._2, true, false) AS _2#5, assertnotnull(input[0, scala.Tuple3, true])._3 AS _3#6] 
//    +- Scan ExternalRDDScan[obj#3] 

Tuttavia altri metodi di KeyValueGroupedDataset potrebbero funzionare in modo simile a RDD.groupByKey. Ad esempio mapGroups (o flatMapGroups) non utilizza l'aggregazione parziale.

ds.groupByKey(_.x) 
    .mapGroups((_, iter) => iter.reduce((x, y) => x.copy(z = x.z + y.z))) 
    .explain 

//== Physical Plan == 
//*SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, $line15.$read$$iw$$iw$Foo, true]).x, true, false) AS x#37, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, $line15.$read$$iw$$iw$Foo, true]).y, true, false) AS y#38, assertnotnull(input[0, $line15.$read$$iw$$iw$Foo, true]).z AS z#39] 
//+- MapGroups <function2>, value#32.toString, newInstance(class $line15.$read$$iw$$iw$Foo), [value#32], [x#8, y#9, z#10], obj#36: $line15.$read$$iw$$iw$Foo 
// +- *Sort [value#32 ASC NULLS FIRST], false, 0 
//  +- Exchange hashpartitioning(value#32, 200) 
//   +- AppendColumns <function1>, newInstance(class $line15.$read$$iw$$iw$Foo), [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true, false) AS value#32] 
//   +- *Project [_1#4 AS x#8, _2#5 AS y#9, _3#6 AS z#10] 
//    +- *SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, scala.Tuple3, true])._1, true, false) AS _1#4, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, scala.Tuple3, true])._2, true, false) AS _2#5, assertnotnull(input[0, scala.Tuple3, true])._3 AS _3#6] 
//     +- Scan ExternalRDDScan[obj#3]