Spark RDD par défaut du nombre de partitions

Version: Spark 1.6.2, Scala 2.10

Je suis de l'exécution de commandes ci-dessous Dans le spark-shell.
Je suis en train de voir le nombre de partitions que l'Étincelle est la création par défaut.

val rdd1 = sc.parallelize(1 to 10)
println(rdd1.getNumPartitions) //==> Result is 4

//Creating rdd for the local file test1.txt. It is not HDFS.
//File content is just one word "Hello"
val rdd2 = sc.textFile("C:/test1.txt")
println(rdd2.getNumPartitions) //==> Result is 2

Que par l'Apache Spark la documentation, le spark.default.parallelism est le nombre de noyaux de mon ordinateur portable (qui est de 2 core du processeur).

Ma question est : rdd2 semblent donner le bon résultat de 2 partitions comme indiqué dans la documentation. Mais pourquoi rdd1 est donner le résultat que 4 partitions ?

J'ai juste observé que quand j'essaie d'exécuter la même(val rdd1 = sc.paralléliser(de 1 à 10)) dans mon Ide, IDE de projet et d'essayer de récupérer le nombre de partiotions, je reçois 2 partitions. Je ne sais pas pourquoi c'est donner un tel résultat dans spark-shell.

OriginalL'auteur Sri | 2017-05-27

  1. 21

    Au minimum le nombre de partitions est en fait une limite inférieure fixée par l' SparkContext. Depuis étincelle utilise hadoop sous le capot, Hadoop InputFormat` sera toujours le comportement par défaut.

    Le premier cas devrait refléter defaultParallelism comme mentionné ici qui peuvent varier en fonction des paramètres et du matériel. (Nombre de cœurs, etc.)

    Donc, sauf si vous fournissez le nombre de tranches, qui premier cas serait défini par le nombre décrite par sc.defaultParallelism:

    scala> sc.defaultParallelism
res0: Int = 6

scala> sc.parallelize(1 to 100).partitions.size
res1: Int = 6

    Comme pour le deuxième cas, avec sc.textFile, le nombre de tranches par défaut est le nombre minimum de partitions.

    Qui est égal à 2 comme vous pouvez le voir dans cette section de code.

    Ainsi, vous devriez considérer les points suivants :

    • sc.parallelize prendra numSlices ou defaultParallelism.

    • sc.textFile va prendre le maximum entre minPartitions et le nombre de divisions calculé sur la base d'hadoop entrée split taille divisée par la taille des blocs.

      • sc.textFile appels sc.hadoopFile, ce qui crée un HadoopRDD qui utilise InputFormat.getSplits sous le capot [Réf. InputFormat documentation].

      • InputSplit[] getSplits(JobConf job, int numSplits) throws IOException : Logiquement divisé l'ensemble des fichiers d'entrée pour le travail.
        Chaque InputSplit est alors attribué à une personne Mappeur pour le traitement.
        Note: Le split est une logique de répartition des intrants et les fichiers d'entrée ne sont pas physiquement divisé en morceaux. Par exemple, le partage pourrait être tuple. Paramètres: emploi - travail de configuration.
        numSplits - le nombre désiré de divisions, d'un soupçon. Retourne: un tableau de InputSplits pour le travail. Jette: IOException.

    Exemple:

    Nous allons créer des mannequin les fichiers de texte:

    fallocate -l 241m bigfile.txt
fallocate -l 4G hugefile.txt

    Cela va créer 2 fichiers, respectivement, de la taille 241MB et de 4 go.

    Nous pouvons voir ce qui se passe quand on lit tous les fichiers: 

    scala> val rdd = sc.textFile("bigfile.txt")
//rdd: org.apache.spark.rdd.RDD[String] = bigfile.txt MapPartitionsRDD[1] at textFile at <console>:27

scala> rdd.getNumPartitions
//res0: Int = 8

scala> val rdd2 = sc.textFile("hugefile.txt")
//rdd2: org.apache.spark.rdd.RDD[String] = hugefile.txt MapPartitionsRDD[3] at textFile at <console>:27

scala> rdd2.getNumPartitions
//res1: Int = 128

    Deux d'entre eux sont en fait HadoopRDDs:

    scala> rdd.toDebugString
//res2: String = 
//(8) bigfile.txt MapPartitionsRDD[1] at textFile at <console>:27 []
// |  bigfile.txt HadoopRDD[0] at textFile at <console>:27 []

scala> rdd2.toDebugString
//res3: String = 
//(128) hugefile.txt MapPartitionsRDD[3] at textFile at <console>:27 []
//  |   hugefile.txt HadoopRDD[2] at textFile at <console>:27 []
    Merci pour votre réponse, mais est ce comportement même lors de la lecture du fichier HDFS à l'aide de sc.fichier texte ? Disons que si je suis à la lecture de la 640 MO fichier HDFS et de saisie de la taille du bloc est de 64 mo.
    texte utilise Hadoop InputFilFormat sous le capot, donc en gros, oui, ça va être la lecture de partitions par bloc d'entrée.
    cette section de code sc.texte utilise defaultMinPartitions qui est calculée conformément à la logique de def defaultMinPartitions: Int = math.min(defaultParallelism, 2). Donc, comme par le code de la sc.fichier texte doit toujours donner le résultat sous la forme 2 partitions quand on ne précise pas le nombre de partitions tout en créant des RDD.
    Je ne suis pas encore clair. Pour les 4 GO de fichier et la taille de bloc de 128 MO. Il aurait dû 4096/128=32 partitions, non? Pourquoi il a donné à 128 partitions. Aussi pourquoi 8 partitions pour 241 MO fichier? Ne devrait-elle pas être de 2 partitions?
    Je suis tombé sur ce thread et a obtenu confondre avec le nombre de partitions créés ici pour une taille de fichier de 241 MO. J'ai recréé les fichiers de telle sorte que chaque taille d'enregistrement est 66KB. Pour une taille de fichier à 255MB, j'ai eu 8 partitions alors que pour une taille de fichier à 531MB, j'ai eu 17 partitions. Je suis incapable de comprendre comment ces partitions valeurs sont-elles décidées?

    OriginalL'auteur eliasah