Почему Spark увеличивает RDD ID на 2 вместо 1 при чтении в текстовых файлах? - PullRequest
Новая
       5

Почему Spark увеличивает RDD ID на 2 вместо 1 при чтении в текстовых файлах?

0 голосов
/ 29 ноября 2018

Я заметил кое-что интересное при работе с spark-shell, и мне интересно, почему это происходит.Я загружаю текстовый файл в Spark, используя основной синтаксис, а затем просто повторяю эту команду.Вывод REPL ниже: 

scala> val myreviews = sc.textFile("Reviews.csv")
myreviews: org.apache.spark.rdd.RDD[String] = Reviews.csv MapPartitionsRDD[1] at textFile at <console>:24

scala> val myreviews = sc.textFile("Reviews.csv")
myreviews: org.apache.spark.rdd.RDD[String] = Reviews.csv MapPartitionsRDD[3] at textFile at <console>:24

scala> val myreviews = sc.textFile("Reviews.csv")
myreviews: org.apache.spark.rdd.RDD[String] = Reviews.csv MapPartitionsRDD[5] at textFile at <console>:24

scala> val myreviews = sc.textFile("Reviews.csv")
myreviews: org.apache.spark.rdd.RDD[String] = Reviews.csv MapPartitionsRDD[7] at textFile at <console>:24

Я знаю, что часть MapPartitionsRDD[X] содержит X в качестве идентификатора СДР.Однако, основываясь на этом сообщении SO об идентификаторах RDD , я ожидаю, что целочисленный идентификатор увеличивается на единицу каждый раз при создании нового RDD.Так почему же он увеличивается на 2?

Я предполагаю, что при загрузке текстового файла создается промежуточный СДР?Потому что создание RDD из parallelize() только увеличивает счетчик RDD на 1 (до 7): 

scala> val arrayrdd = sc.parallelize(Array(3,4,5))
arrayrdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[8] at parallelize at <console>:24

Примечание: я не верю, что число имеет отношение к разделам.Если я позвоню, я получу, что мой RDD разделен на 9 разделов: 

scala> myreviews.partitions.size
res2: Int = 9

1 Ответ

0 голосов
/ 29 ноября 2018

Поскольку один вызов метода может создать более одного промежуточного RDD.Это будет очевидно, если вы проверите строку отладки 

sc.textFile("README.md").toDebugString

String =
(2) README.md MapPartitionsRDD[1] at textFile at <console>:25 []
 |  README.md HadoopRDD[0] at textFile at <console>:25 []

Как видите, линия состоит из двух RDD.

  • Первая - HadoopRDD, что соответствует импорту данных 

    hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],
  minPartitions)

  • Второй - MapPartitionsRDD и соответствует последующим map, которыесбрасывает ключи (смещения) и преобразует Text в String. 

    .map(pair => pair._2.toString).setName(path)
...