Итерации по СДР Итерируемые в Scala - PullRequest
Новая
       22

Итерации по СДР Итерируемые в Scala

0 голосов
/ 01 февраля 2019

Итак, я новичок в Scala и только начинаю работать с RDD и функциональными операциями Scala.

Я пытаюсь перебрать значения моих парных СДР и вернуть Var1 со средним значением, сохраненным в Var2, применяя определенную функцию average, чтобы окончательный результат был уникальнымсписок Var1 с одним AvgVar2, связанным с каждым.У меня много проблем с выяснением, как перебирать значения.

* edit: у меня есть следующие объявления типов: 

case class ID: Int,  Var1: Int, Var2: Int extends Serializable

У меня есть следующая функция: 

  def foo(rdds: RDD[(ID, Iterable[(Var1, Var2)])]): RDD[(Var1, AvgVar2)] = {

    def average(as: Array[Var2]): AvgVar2 = {
       var sum = 0.0
       var i = 0.0
       while (i < as.length) {
           sum += Var2.val
           i += 1
      }
      sum/i
    }

    //My attempt at Scala
    rdds.map(x=> ((x._1),x._2)).groupByKey().map(x=>average(x._1)).collect()
}

Моя попытка Scala пытается сделать следующее:

  1. разбить итерабельную пару СДР на пары ключ-значение Var1-Var2.
  2. Группировать по ключу Var1 и создать массив связанных Var2.
  3. Применить мою функцию average к каждому массиву Var2
  4. Возвращать AvgVar2 с ассоциированным Var1 в виде набора СДР

*Редактировать:

Некоторые примеры входных данных для rdds: 

//RDD[(ID,Iterable[(Var1,Var2)...])]
RDD[(1,[(1,3),(1,12),(1,6)])],
RDD[(2,[(2,5),(2,7)])]

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

//RDD[(Var1, AvgVar2)]
RDD[(1,7),(2,6)]

* Редактировать: строка рабочего скалярного кода: 

rdd.map(x => (x._2.map(it => it._1).asInstanceOf[Var1], average(x._2.map(it => it._2).toArray)))

1 Ответ

0 голосов
/ 01 февраля 2019

Учитывая ID = Var1, простое .map() решит это: 

def foo(rdds: RDD[(Int, Iterable[(Int, Int)])]): RDD[(Int, Double)] = {

  def average(as: Iterable[(Int, Int)]): Double = {
    as.map(_._2).reduce(_+_)/as.size.toDouble
  }

  rdds.map(x => (x._1, average(x._2)))
}

Вывод: 

val input = sc.parallelize(List((1,Iterable((1,3),(1,12),(1,6))), (2, Iterable((2,5),(2,7)))))

scala> foo(input).collect
res0: Array[(Int, Double)] = Array((1,7.0), (2,6.0))

РЕДАКТИРОВАНИЕ: (average() с такой же подписью): 

def foo(rdds: RDD[(Int, Iterable[(Int, Int)])]): RDD[(Int, Double)] = {

  def average(as: Array[Int]): Double = {
    as.reduce(_+_)/as.size.toDouble
  }

  rdds.map(x => (x._1, average(x._2.map(tuple => tuple._2).toArray)))
}
...