Уменьшить хэш-карты на раздел в Spark - PullRequest
Новая
       9

Уменьшить хэш-карты на раздел в Spark

0 голосов
/ 24 июня 2018

У меня есть RDD некоторого mutable.Map[(Int, Array[Double])], и я хотел бы уменьшить карты на Int и найти средства для элементов массивов.

Например, у меня есть: 

Map[(1, Array[0.1, 0.1]), (2, Array[0.3, 0.2])] 
Map[(1, Array[0.1, 0.4])]

Что я хочу: 

Map[(1, Array[0.1, 0.25]), (2, Array[0.3, 0.2])]

Проблема в том, что я не знаю, как reduce работает между картами иКроме того, я должен сделать это для каждого раздела, собрать результаты в драйвер и уменьшить их там тоже.Я нашел метод foreachPartition, но я не знаю, предназначен ли он для использования в таких случаях.

Есть идеи?

1 Ответ

0 голосов
/ 24 июня 2018

Вы можете сделать это, используя combineByKey: 

val rdd = ss.sparkContext.parallelize(Seq(
  Map((1, Array(0.1, 0.1)), (2, Array(0.3, 0.2))),
  Map((1, Array(0.1, 0.4)))
))

// functions for combineByKey
val create = (arr: Array[Double]) => arr.map( x => (x,1))
val update = (acc : Array[(Double,Int)], current: Array[Double]) => acc.zip(current).map{case ((s,c),x) => (s+x,c+1)}
val merge =  (acc1 : Array[(Double,Int)],acc2:Array[(Double,Int)]) => acc1.zip(acc2).map{case ((s1,c1),(s2,c2)) => (s1+s2,c1+c2)}

val finalMap = rdd.flatMap(_.toList)
  // aggreate elementwise sum & count
  .combineByKey(create,update,merge)
  // calculate elementwise average per key
  .map{case (id,arr) => (id,arr.map{case (s,c) => s/c})}
  .collectAsMap()

// finalMap = Map(2 -> Array(0.3, 0.2), 1 -> Array(0.1, 0.25))
...