Прежде всего, избегайте 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