Метод aggregateByKey не работает в спарк rdd - PullRequest
Новая
       29

Метод aggregateByKey не работает в спарк rdd

0 голосов
/ 06 октября 2018

Ниже приведены мои примеры данных: 

1,Siddhesh,43,32000
1,Siddhesh,12,4300
2,Devil,10,1000
2,Devil,10,3000
2,Devil,11,2000

Я создал пару СДР для выполнения операций combineByKey и aggregateByKey.Ниже мой код: 

val rd=sc.textFile("file:///home/cloudera/Desktop/details.txt").map(line=>line.split(",")).map(p=>((p(0).toString,p(1).toString),(p(3).toLong,p(2).toString.toInt)))

Выше я спарил данные первых двух столбцов в качестве ключа и оставшихся столбцов в качестве значения.Теперь я хочу только отличные значения от правого кортежа для 3-го столбца в наборе данных, который я смог сделать с combByKey.Ниже приведен мой код: 

val reduced = rd.combineByKey(
      (x:(Long,Int))=>{(x._1,Set(x._2))},
      (x:(Long,Set[Int]),y:(Long,Int))=>(x._1+y._1,x._2+y._2),
      (x:(Long,Set[Int]),y:(Long,Set[Int]))=>{(x._1+y._1,x._2++y._2)}
      )  
scala> reduced.foreach(println)
((1,Siddhesh),(36300,Set(43, 12)))
((2,Devil),(6000,Set(10, 11)))

Теперь я сопоставляю его, чтобы получить сумму значений уникальных отдельных ключей.

scala> val newRdd=reduced.map(p=>(p._1._1,p._1._2,p._2._1,p._2._2.size))

scala> newRdd.foreach(println)
(1,Siddhesh,36300,2)
(2,Devil,6000,2)

Здесь для дьявола последнее значение равно 2, так как у меня есть 10 как 2 значения для записи «Дьявол» в наборе данных, и поскольку я использовал Set, он удаляет дубликаты.Так что теперь я попробовал это с aggregateByKey.Ниже приведен мой код с ошибкой: 

val rd=sc.textFile("file:///home/cloudera/Desktop/details.txt").map(line=>line.split(",")).map(p=>((p(0).toString,p(1).toString),(p(3).toString.toInt,p(2).toString.toInt)))

Я преобразовал столбец значений из long в int, потому что во время инициализации он выдавал ошибку на «0» 

scala> val reducedByAggKey = rd.aggregateByKey((0,0))(
     |        (x:(Int,Set[Int]),y:(Int,Int))=>(x._1+y._1,x._2+y._2),
     |       (x:(Int,Set[Int]),y:(Int,Set[Int]))=>{(x._1+y._1,x._2++y._2)}
     | )
<console>:36: error: type mismatch;
 found   : scala.collection.immutable.Set[Int]
 required: Int
              (x:(Int,Set[Int]),y:(Int,Int))=>(x._1+y._1,x._2+y._2),
                                                             ^
<console>:37: error: type mismatch;
 found   : scala.collection.immutable.Set[Int]
 required: Int
             (x:(Int,Set[Int]),y:(Int,Set[Int]))=>{(x._1+y._1,x._2++y._2)}
                                                                  ^

И, как предложил Леониже приведен мой код с ошибкой: 

    scala> val reduced = rdd.aggregateByKey((0, Set.empty[Int]))(
     |   (x: (Int, Set[Int]), y: (Int, Int)) => (x._1 + y._1, y._2+x._2),
     |   (x: (Int, Set[Int]), y: (Int, Set[Int])) => (x._1 + y._1, y._2++ x._2)
     | )
<console>:36: error: overloaded method value + with alternatives:
  (x: Double)Double <and>
  (x: Float)Float <and>
  (x: Long)Long <and>
  (x: Int)Int <and>
  (x: Char)Int <and>
  (x: Short)Int <and>
  (x: Byte)Int <and>
  (x: String)String
 cannot be applied to (Set[Int])
         (x: (Int, Set[Int]), y: (Int, Int)) => (x._1 + y._1, y._2+x._2),
                                                                  ^

Так, где я тут путаюсь ??Пожалуйста, поправьте меня

1 Ответ

0 голосов
/ 06 октября 2018

Если я правильно понимаю ваше требование, чтобы получить полный счет, а не отдельный счет, используйте List вместо Set для агрегатов.Что касается проблемы с вашим aggregateByKey, это связано с неправильным типом zeroValue, который должен быть (0, List.empty[Int]) (был бы (0, Set.empty[Int]), если бы вы использовали Set): 

val reduced = rdd.aggregateByKey((0, List.empty[Int]))(
  (x: (Int, List[Int]), y: (Int, Int)) => (x._1 + y._1, y._2 :: x._2),
  (x: (Int, List[Int]), y: (Int, List[Int])) => (x._1 + y._1, y._2 ::: x._2)
)

reduced.collect
// res1: Array[((String, String), (Int, List[Int]))] =
//   Array(((2,Devil),(6000,List(11, 10, 10))), ((1,Siddhesh),(36300,List(12, 43))))

val newRdd = reduced.map(p => (p._1._1, p._1._2, p._2._1, p._2._2.size))

newRdd.collect
// res2: Array[(String, String, Int, Int)] =
//   Array((2,Devil,6000,3), (1,Siddhesh,36300,2))

Обратите внимание, что изменение Set на List будет применяться и к вашему коду combineByKey, если вы хотите использовать полный счет вместо отдельного подсчета.

[ОБНОВЛЕНИЕ]

Для точного подсчета за ваш комментарий просто оставайтесь с Set с zeroValue, установленным на (0, Set.empty[Int]): 

val reduced = rdd.aggregateByKey((0, Set.empty[Int]))(
  (x: (Int, Set[Int]), y: (Int, Int)) => (x._1 + y._1, x._2 + y._2),
  (x: (Int, Set[Int]), y: (Int, Set[Int])) => (x._1 + y._1, x._2 ++ y._2)
)

reduced.collect
// res3: Array[((String, String), (Int, scala.collection.immutable.Set[Int]))] =
//   Array(((2,Devil),(6000,Set(10, 11))), ((1,Siddhesh),(36300,Set(43, 12))))
...