Key-Value RDD: почему одинаковые дубликаты значений? (Скала)

Question

Я очень новичок в Scala, и при создании приложения я столкнулся с этим странным эффектом:

Сначала позвольте мне рассказать вам несколько деталей:

У меня есть набор данных, в который я импортируюмое приложение и делать вещи. Это CSV, в котором каждая строка имеет такую ​​форму:

x : Double, y : Double, class : String

Каждая строка этого набора данных содержит координаты точки и класс этой точки.

Моя цель, после некоторых вычислений, вычислить новое значение на основе этого набора данных и создать СДР, используя данный набор данных в сочетании с новым столбцом, содержащим новое значение, с ключом-значением пара.

СДР будет иметь в качестве ключа координаты (x, y) каждой точки и в качестве значения класса И новое значение с именем icch_id:Строка!

Правильная форма для этого СДР, на мой взгляд, будет включать в себя два кортежа, которые будут выглядеть следующим образом:

(x : Double, y : Double),(icch_id : String, class : String)

Я достигаю точки в моем приложении после некоторых вычислений, где у меня естьэтот СДР для работы с:

DATA_RDD:

(icch_id : String, x : Double, y : Double, class : String) 

Печать части этого СДР показана ниже:

//DATA_RDD print:
(ICCH_1,4.3,3.0,Iris-setosa)
(ICCH_1,4.4,2.9,Iris-setosa)

И я пишу этот код, чтобы преобразовать его в нужную мне форму ключа-значения, которую я написал выше:

val DATA_RDD_Final  = DATA_RDD
      .keyBy{case (_, x, y, _) => (x, y)}
      .mapValues{ case (icch_id, _, _, pclass) => (icch_id,pclass) }
      .groupByKey()

И на этом этапе начинается проблема.

Если я печатаю DATA_RDD_Final , результат будет:

//A part of the Result:
((5.4,3.0),CompactBuffer((ICCH_1,Iris-versicolor))) //Almost Correct!
((6.7,3.3),CompactBuffer((ICCH_3,Iris-virginica), (ICCH_3,Iris-virginica))) //Wrong!

Проблема: Вместо простого кортежа я получаю CompactBuffer с точно таким жедубликаты Value Tuple.

И я спрашиваю:

1. Почему это происходит? Это проблема кода?
2. Я использую редакцию сообщества databricks. Это как-то влияет?
3. Есть ли способ, после того, как я создаю DATA_RDD_Final, чтобы удалить все дубликаты и вернуть форму, которую я хотел с самого начала?

Большое вам спасибо заот имени и извините за длинный пост!

Answer 1

Вы можете достичь этого с помощью одного map:

rdd.map {
  case (id, x, y, pclass) => ((x, y), (id, pclass))
}

Answer 2

Хорошо, я нашел решение!

Использование .groupByKey() создает проблему. Когда я удаляю его и просто использую .collect()

val Data_Array_Final  = Data_RDD
      .keyBy{case (_, x, y, _) => (x, y)}
      .mapValues{ case (id, _, _, pclass) => (id,pclass) }
      .collect()

Все выглядит хорошо, за исключением одного небольшого недостатка. Вместо RDD он возвращает массив. Поэтому мне нужно было сделать:

val Data_RDD_Final = sc.parallelize(Data_Array_Final)

, чтобы получить желаемый результат. Теперь у меня есть СДР правильной формы:

org.apache.spark.rdd.RDD[((Double, Double), (String, String))]