Совокупные значения RDD для каждого ключа

У меня есть RDD в структуре ключа, значения (someKey, (мера1, мера2)). Я сгруппировал по ключу, и теперь я хочу агрегировать значения для каждого ключа.

val RDD1 : RDD[(String,(Int,Int))]
RDD1.groupByKey()

результат, который мне нужен:

key: avg(measure1), avg(measure2), max(measure1), max(measure2), min(measure1), min(measure2), count(*)

apache-spark scala rdd aggregate-functions
person Userrrrrrrr    schedule 20.04.2015    source источник

Ответы (1)


arrow_upward
5
arrow_downward

Прежде всего, избегайте groupByKey! Вы должны использовать aggregateByKey или combineByKey. Мы будем использовать aggregateByKey< /а>. Эта функция преобразует значения для каждого ключа: RDD[(K, V)] => RDD[(K, U)]. Ему нужно нулевое значение типа U и знание того, как объединить (V, U) => U и (U, U) => U. Я немного упростил ваш пример и хочу получить: key: avg(measure1), avg(measure2), min(measure1), min(measure2), count(*)

  val rdd1 = sc.parallelize(List(("a", (11, 1)), ("a",(12, 3)), ("b",(10, 1))))
  rdd1
    .aggregateByKey((0.0, 0.0, Int.MaxValue, Int.MaxValue, 0))(
      {
        case ((sum1, sum2, min1, min2, count1), (v1, v2)) =>
          (sum1 + v1, sum2 + v2, v1 min min1, v2 min min2, count1+1)
      }, 
      {
        case ((sum1, sum2, min1, min2, count),
          (otherSum1, otherSum2, otherMin1, otherMin2, otherCount)) =>
          (sum1 + otherSum1, sum2 + otherSum2, 
           min1 min otherMin1, min2 min otherMin2, count + otherCount)
      }
    )
    .map {
      case (k, (sum1, sum2, min1, min2, count1)) => (k, (sum1/count1, sum2/count1, min1, min2, count1))
    }
    .collect()

давать

   (a,(11.5,2.0,11,1,2)), (b,(10.0,1.0,10,1,1))
person Nikita    schedule 20.04.2015