Comment définir le partitionnement de DataFrame?

Étincelle >= 2.3.0

SPARK-22614 expose le partitionnement de plage.

val partitionedByRange = df.repartitionByRange(42, $"k")

partitionedByRange.explain
// == Parsed Logical Plan ==
// 'RepartitionByExpression ['k ASC NULLS FIRST], 42
// +- AnalysisBarrier Project [_1#2 AS k#5, _2#3 AS v#6]
// 
// == Analyzed Logical Plan ==
// k: string, v: int
// RepartitionByExpression [k#5 ASC NULLS FIRST], 42
// +- Project [_1#2 AS k#5, _2#3 AS v#6]
//    +- LocalRelation [_1#2, _2#3]
// 
// == Optimized Logical Plan ==
// RepartitionByExpression [k#5 ASC NULLS FIRST], 42
// +- LocalRelation [k#5, v#6]
// 
// == Physical Plan ==
// Exchange rangepartitioning(k#5 ASC NULLS FIRST, 42)
// +- LocalTableScan [k#5, v#6]

SPARK-22389 expose le partitionnement de format externe dans l'API de source de données V2 .

Étincelle >= 1.6.0

Dans Spark > = 1.6, il est possible d'utiliser le partitionnement par colonne pour la requête et la mise en cache. Voir: SPARK-11410 et SPARK-4849 utiliser repartition méthode:

val df = Seq(
  ("A", 1), ("B", 2), ("A", 3), ("C", 1)
).toDF("k", "v")

val partitioned = df.repartition($"k")
partitioned.explain

// scala> df.repartition($"k").explain(true)
// == Parsed Logical Plan ==
// 'RepartitionByExpression ['k], None
// +- Project [_1#5 AS k#7,_2#6 AS v#8]
//    +- LogicalRDD [_1#5,_2#6], MapPartitionsRDD[3] at rddToDataFrameHolder at <console>:27
// 
// == Analyzed Logical Plan ==
// k: string, v: int
// RepartitionByExpression [k#7], None
// +- Project [_1#5 AS k#7,_2#6 AS v#8]
//    +- LogicalRDD [_1#5,_2#6], MapPartitionsRDD[3] at rddToDataFrameHolder at <console>:27
// 
// == Optimized Logical Plan ==
// RepartitionByExpression [k#7], None
// +- Project [_1#5 AS k#7,_2#6 AS v#8]
//    +- LogicalRDD [_1#5,_2#6], MapPartitionsRDD[3] at rddToDataFrameHolder at <console>:27
// 
// == Physical Plan ==
// TungstenExchange hashpartitioning(k#7,200), None
// +- Project [_1#5 AS k#7,_2#6 AS v#8]
//    +- Scan PhysicalRDD[_1#5,_2#6]

Contrairement à RDDs Spark Dataset (y compris Dataset[Row] alias DataFrame) ne peut pas utilisez le partitionneur personnalisé comme pour l'instant. Vous pouvez généralement résoudre ce problème en créant une colonne de partitionnement artificiel, mais cela ne vous donnera pas la même flexibilité.

Étincelle

Une chose que vous pouvez faire est de pré-partitionner les données d'entrée avant de créer un DataFrame

import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
import org.apache.spark.HashPartitioner

val schema = StructType(Seq(
  StructField("x", StringType, false),
  StructField("y", LongType, false),
  StructField("z", DoubleType, false)
))

val rdd = sc.parallelize(Seq(
  Row("foo", 1L, 0.5), Row("bar", 0L, 0.0), Row("??", -1L, 2.0),
  Row("foo", -1L, 0.0), Row("??", 3L, 0.6), Row("bar", -3L, 0.99)
))

val partitioner = new HashPartitioner(5) 

val partitioned = rdd.map(r => (r.getString(0), r))
  .partitionBy(partitioner)
  .values

val df = sqlContext.createDataFrame(partitioned, schema)

Puisque DataFrame la création à partir d'un RDD ne nécessite qu'une simple phase de map, la disposition de partition existante doit être préservée*:

assert(df.rdd.partitions == partitioned.partitions)

De la même manière que vous pouvez repartitionner existant DataFrame:

sqlContext.createDataFrame(
  df.rdd.map(r => (r.getInt(1), r)).partitionBy(partitioner).values,
  df.schema
)

Il semble donc que ce n'est pas impossible. La question reste de savoir si cela a du sens. Je dirais que la plupart du temps, il ne le fait pas:

1. Le repartitionnement est un processus coûteux. Dans un scénario typique, la plupart des données doivent être sérialisées, mélangées et désérialisées. D'autre part, le nombre d'opérations pouvant bénéficier d'une donnée pré-partitionnée est relativement faible et est encore limité si L'API interne n'est pas conçue pour tirer parti de cela propriété.

   • se joint à certains scénarios, mais cela nécessiterait un support interne,
   • les fonctions de fenêtre appellent avec le partitionneur correspondant. Comme ci-dessus, limité à une définition de guichet unique. Il est déjà partitionné en interne, donc le pré-partitionnement peut être redondant,
   • agrégations simples avec GROUP BY - Il est possible de réduire l'empreinte mémoire des tampons temporaires**, mais le coût global est beaucoup plus élevé. Plus ou moins équivalent à groupByKey.mapValues(_.reduce) (comportement actuel) vs reduceByKey (pré-partitionnement). Peu probable d'être utile dans la pratique.
   • compression de données avec SqlContext.cacheTable. Comme il semble qu'il utilise l'encodage de longueur d'exécution, l'application de OrderedRDDFunctions.repartitionAndSortWithinPartitions pourrait améliorer le taux de compression.

2. Les performances dépendent fortement d'une distribution des clés. S'il est biaisé, il en résultera une utilisation des ressources sous-optimale. Dans le pire des cas, il sera impossible de terminer le travail à tous.

3. tout un point d'utiliser un haut L'API déclarative de niveau est de vous isoler d'un détail d'implémentation de bas niveau. Comme déjà mentionné par @dwysakowicz et @RomiKuntsman, une optimisation est un travail de l' Catalyseur Optimiseur de. C'est une bête assez sophistiquée et je doute vraiment que vous puissiez facilement améliorer cela sans plonger beaucoup plus profondément dans ses internes.

Concepts connexes

Partitionnement avec les sources JDBC :

Prise en charge des sources de données JDBC predicates l'argument. Il peut être utilisé comme suit:

sqlContext.read.jdbc(url, table, Array("foo = 1", "foo = 3"), props)

Il crée une seule partition JDBC par prédicat. Gardez à l'esprit que si les ensembles créés à l'aide de prédicats individuels ne sont pas disjoints, vous verrez des doublons dans la table résultante.

partitionBy méthode DataFrameWriter:

Spark DataFrameWriter fournit la méthode partitionBy qui peut être utilisée pour "partitionner" les données en écriture. Il sépare les données en écriture en utilisant l'ensemble de colonnes fourni

val df = Seq(
  ("foo", 1.0), ("bar", 2.0), ("foo", 1.5), ("bar", 2.6)
).toDF("k", "v")

df.write.partitionBy("k").json("/tmp/foo.json")

Cela permet prédicat pousser vers le bas en lecture pour les requêtes basées sur la clé:

val df1 = sqlContext.read.schema(df.schema).json("/tmp/foo.json")
df1.where($"k" === "bar")

Mais ce n'est pas équivalent à DataFrame.repartition. En particulier les agrégations comme:

val cnts = df1.groupBy($"k").sum()

Nécessitera toujours TungstenExchange:

cnts.explain

// == Physical Plan ==
// TungstenAggregate(key=[k#90], functions=[(sum(v#91),mode=Final,isDistinct=false)], output=[k#90,sum(v)#93])
// +- TungstenExchange hashpartitioning(k#90,200), None
//    +- TungstenAggregate(key=[k#90], functions=[(sum(v#91),mode=Partial,isDistinct=false)], output=[k#90,sum#99])
//       +- Scan JSONRelation[k#90,v#91] InputPaths: file:/tmp/foo.json

bucketBy méthode DataFrameWriter (Spark >= 2.0):

bucketBy a des applications similaires à partitionBy mais il n'est disponible que pour les tables (saveAsTable). Écopage informations peuvent être utilisées pour optimiser les jointures:

// Temporarily disable broadcast joins
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)

df.write.bucketBy(42, "k").saveAsTable("df1")
val df2 = Seq(("A", -1.0), ("B", 2.0)).toDF("k", "v2")
df2.write.bucketBy(42, "k").saveAsTable("df2")

// == Physical Plan ==
// *Project [k#41, v#42, v2#47]
// +- *SortMergeJoin [k#41], [k#46], Inner
//    :- *Sort [k#41 ASC NULLS FIRST], false, 0
//    :  +- *Project [k#41, v#42]
//    :     +- *Filter isnotnull(k#41)
//    :        +- *FileScan parquet default.df1[k#41,v#42] Batched: true, Format: Parquet, Location: InMemoryFileIndex[file:/spark-warehouse/df1], PartitionFilters: [], PushedFilters: [IsNotNull(k)], ReadSchema: struct<k:string,v:int>
//    +- *Sort [k#46 ASC NULLS FIRST], false, 0
//       +- *Project [k#46, v2#47]
//          +- *Filter isnotnull(k#46)
//             +- *FileScan parquet default.df2[k#46,v2#47] Batched: true, Format: Parquet, Location: InMemoryFileIndex[file:/spark-warehouse/df2], PartitionFilters: [], PushedFilters: [IsNotNull(k)], ReadSchema: struct<k:string,v2:double>

* Par disposition de partition je veux dire seulement de données distribution. partitioned RDD n'a plus de partitionneur. ** En supposant qu'il n'y ait pas de projection anticipée. Si l'agrégation ne couvre qu'un petit sous-ensemble de colonnes, il n'y a probablement aucun gain.