Obtenir le nombre actuel de partitions d'un DataFrame

Question

Existe-t-il un moyen d'obtenir le nombre actuel de partitions d'un DataFrame ? J'ai vérifié la javadoc de DataFrame (spark 1.6) et je n'ai pas trouvé de méthode pour cela, ou est-ce que je l'ai juste manqué ? (Dans le cas de JavaRDD, il y a une méthode getNumPartitions()).

Answer 1

Vous devez appeler getNumPartitions() sur le RDD sous-jacent du DataFrame, par exemple, df.rdd.getNumPartitions() . Dans le cas de Scala, il s'agit d'une méthode sans paramètre : df.rdd.getNumPartitions .

Answer 2

dataframe.rdd.partitions.size est une autre alternative en dehors de df.rdd.getNumPartitions() o df.rdd.length .

Laissez-moi vous expliquer cela avec un exemple complet...

val x = (1 to 10).toList
val numberDF = x.toDF(“number”)
numberDF.rdd.partitions.size // => 4

Pour prouver que le nombre de partitions que nous avons obtenu avec ci-dessus ... enregistrer ce cadre de données comme csv

numberDF.write.csv(“/Users/Ram.Ghadiyaram/output/numbers”)

Voici comment les données sont séparées sur les différentes partitions.

Partition 00000: 1, 2
Partition 00001: 3, 4, 5
Partition 00002: 6, 7
Partition 00003: 8, 9, 10

Mise à jour :

@Hemanth a posé une bonne question dans le commentaire ... fondamentalement pourquoi le nombre de partitions est de 4 dans le cas ci-dessus

Réponse courte : Cela dépend des cas où vous exécutez. Comme j'ai utilisé local[4], j'ai obtenu 4 partitions.

Réponse longue :

J'exécutais le programme ci-dessus dans ma machine locale et j'ai utilisé master comme local[4] en fonction de cela il prenait comme 4 les partitions.

val spark = SparkSession.builder()
    .appName(this.getClass.getName)
    .config("spark.master", "local[4]").getOrCreate()

Si c'est spark-shell dans le fil principal, le nombre de partitions est de 2.

exemple : spark-shell --master yarn et j'ai tapé les mêmes commandes à nouveau

scala> val x = (1 to 10).toList
x: List[Int] = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

scala> val numberDF = x.toDF("number")
numberDF: org.apache.spark.sql.DataFrame = [number: int]

scala> numberDF.rdd.partitions.size
res0: Int = 2

• ici 2 est parllelisme par défaut de spark
• Sur la base de hashpartitioner spark décidera du nombre de partitions à distribuer. si vous exécutez en --master local et en fonction de votre Runtime.getRuntime.availableProcessors() c'est-à-dire local[Runtime.getRuntime.availableProcessors()] il va essayer d'allouer Si le nombre de processeurs dont vous disposez est de 12 (c'est à dire local[Runtime.getRuntime.availableProcessors()]) et vous avez une liste de 1 à 10 alors seulement 10 partitions seront créées.

NOTA:

*_Si vous êtes sur un ordinateur portable à 12 cœurs où j'exécute le programme spark et que par défaut le nombre de partitions/tâches est le nombre de tous les cœurs disponibles, soit 12, cela signifie que signifie `local[]ous"local[${Runtime.getRuntime.availableProcessors()}]")` mais dans ce mais dans ce cas, il n'y a que 10 numéros, donc il se limitera à 10._**

En gardant tous ces points à l'esprit, je vous suggère d'essayer par vous-même

Answer 3

Convertir en RDD puis obtenir la longueur des partitions

DF.rdd.partitions.length

Answer 4

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

df.rdd.getNumPartitions

Answer 5

Une autre façon intéressante d'obtenir le nombre de partitions est d'utiliser la transformation 'mapPartitions'. Exemple de code -

val x = (1 to 10).toList
val numberDF = x.toDF()
numberDF.rdd.mapPartitions(x => Iterator[Int](1)).sum()

Les experts de Spark sont invités à commenter ses performances.