Скала повторной сортировки в результате

Question

У меня есть СДР, в котором указано "Клиент", "Потрачено". Я пытаюсь выполнить простую сортировку по заказу от "Amt Spent". Когда я просматриваю результаты, я вижу, что происходит множественная сортировка.

• Мой код:

  val sc = new SparkContext("local[*]", "TotalSpentByCustomerSorted")   
  
  val input = sc.textFile("filename.csv")
  
  val input2 = input.map(line=>line.split(",")).map(
    line=>(line(0).toInt,line(2).toFloat)
  ) // To fetch only customer#, Amt spent
  
  val input3 = input2.reduceByKey((x,y)=>(x+y)).sortBy(x=>x._2)
  
  input3.foreach(println)
  

• Результат:

  (28,5000.7104)
  (22,5019.449)
  (29,5032.5303)
  (17,5032.6797)
  (60,5040.7095)
  (45,3309.3804)
  (79,3790.5698)
  (96,3924.23)
  (23,4042.65)
  (99,4172.29)
  

Как видно из результата, первые 5 записей отсортированы по потраченной сумме, а затем снова происходит другая сортировка. В чем может быть проблема здесь?

Answer 1

Проблема не в том, что сортировка не работала, а в том, что вы println() вызывали foreach на СДР - эта операция выполняется параллельно на всех разделах.
И на реальном кластере (не в локальной среде разработки) вы даже не увидите напечатанные строки, потому что это произойдет на JVM исполнителей .

RDD отсортировано, но если вы не хотите верить мне (это было бы разумно) , тогда вы можете выполнить collect до foreach , который сначала извлечет все результаты для драйвера, а затем напечатает их (он извлечет их в том порядке, в котором они находятся) .