scala - 计算 Hadoop 上偶数/奇数对的总和？

Question

我想为 Hadoop 创建一个并行 scanLeft(计算关联运算符的前缀和)函数(特别是烫伤；请参阅下文了解如何完成)。

给定一个 hdfs 文件中的数字序列(每行一个)，我想用连续偶数/奇数对的总和计算一个新序列。例如:

输入序列:

0,1,2,3,4,5,6,7,8,9,10

输出序列:

0+1、2+3、4+5、6+7、8+9、10

即

1,5,9,13,17,10

我认为，为了做到这一点，我需要为 Hadoop 编写一个 InputFormat 和 InputSplits 类，但我不知道该怎么做。

参见本节 3.3 here .以下是 Scala 中的示例算法:

 // for simplicity assume input length is a power of 2

def scanadd(input : IndexedSeq[Int]) : IndexedSeq[Int] =   
if (input.length == 1)
  input 
else { 
//calculate a new collapsed sequence which is the sum of sequential even/odd pairs 
val collapsed = IndexedSeq.tabulate(input.length/2)(i => input(2 * i) + input(2*i+1))

//recursively scan collapsed values
val scancollapse = scanadd(collapse)

//now we can use the scan of the collapsed seq to calculate the full sequence

val output = IndexedSeq.tabulate(input.length)(
i => i.evenOdd match {             

//if an index is even then we can just look into the collapsed sequence and get the value
// otherwise we can look just before it and add the value at the current index

   case Even => scancollapse(i/2) 
   case Odd => scancollapse((i-1)/2) + input(i)  
}

output
}

我知道这可能需要相当多的优化才能与 Hadoop 很好地协同工作。直接翻译这个我认为会导致 Hadoop 代码非常低效。例如，显然在 Hadoop 中您不能使用 IndexedSeq。如果您看到任何具体问题，我将不胜感激。不过，我认为它可能会运作良好。

Answer 1

多余的。你是说这段代码？

val vv = (0 to 1000000).grouped(2).toVector
vv.par.foldLeft((0L, 0L, false))((a, v) => 
    if (a._3) (a._1, a._2 + v.sum, !a._3) else (a._1 + v.sum, a._2, !a._3))