Grupo DataFrame / Conjunto de Dados Por comportamento / otimização

Suponha que temos o DataFrame df consistindo nas seguintes colunas:

Nome, sobrenome, tamanho, largura, comprimento, pesar

Agora queremos realizar algumas operações, por exemplo, queremos criar alguns DataFrames contendo dados sobre Tamanho e Largura.

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

como você pode notar, outras colunas, como Comprimento, não são usadas em nenhum lugar. O Spark é inteligente o suficiente para descartar as colunas redundantes antes da fase de embaralhamento ou elas são carregadas? Wil correndo:

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

antes de agrupar de alguma forma afetar o desempenho?

Sim, é ” inteligente o suficiente “. groupBy realizado em um DataFrame não é a mesma operação que groupBy executou em um RDD simples. Em um cenário que você descreveu, não há necessidade de mover dados brutos. Vamos criar um pequeno exemplo para ilustrar isso:

 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] 

Como você pode, a primeira fase é uma projeção onde apenas as colunas necessárias são preservadas. Os próximos dados são agregados localmente e finalmente transferidos e agregados globalmente. Você obterá uma saída de resposta um pouco diferente se usar Spark <= 1.4, mas a estrutura geral deve ser exatamente a mesma.

Finalmente, uma visualização do DAG mostrando que a descrição acima descreve o trabalho real:

Agrupar por e agg DAG

Da mesma forma, Dataset.groupByKey seguido por reduceGroups , contém os dois lados do mapa ( ObjectHashAggregate com partial_reduceaggregator ) e reduce-side ( ObjectHashAggregate com reduceaggregator reduction):

 case class Foo(x: String, y: String, z: Int) val ds = df.as[Foo] ds.groupByKey(_.x).reduceGroups((x, y) => x.copy(z = xz + yz)).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 , 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] 

groupByKey + reduceGroups

No entanto, outros methods de KeyValueGroupedDataset podem funcionar de maneira semelhante a RDD.groupByKey . Por exemplo, mapGroups (ou flatMapGroups ) não usa agregação parcial.

 ds.groupByKey(_.x) .mapGroups((_, iter) => iter.reduce((x, y) => x.copy(z = xz + yz))) .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 , 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 , 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] 

groupByKey + mapGroups