如何使用Spark计算累计和

Question

我有一个rdd（String，Int），它是按键排序的

val data = Array(("c1",6), ("c2",3),("c3",4))
val rdd = sc.parallelize(data).sortByKey

现在，我想用零开始第一个键的值，并将后续键作为前一个键的和。

例如：c1=0，c2=c1的值，c3=（c1值c2值），c4=（c1..c3值）预期输出：

(c1,0), (c2,6), (c3,9)...

有可能做到这一点吗？我用地图试过了，但总和没有保存在地图里。

var sum = 0 ;
val t = keycount.map{ x => { val temp = sum; sum = sum + x._2 ; (x._1,temp); }}

Answer 1

我遇到了一个类似的问题，并实现了保罗的解决方案。我想做一个整数频率表按键排序（整数），有一个小问题与np.cumsum（partition_sums），错误是不支持的操作数类型（s）=：'int'和'NoneType' .

因为如果范围足够大，那么每个分区拥有某样东西的概率就足够大（没有无值）。但是，如果范围远小于计数，并且分区数量保持不变，则某些分区将为空。修改后的解决方案来了：

def cumsum(rdd, get_summand):
    """Given an ordered rdd of items, computes cumulative sum of
    get_summand(row), where row is an item in the RDD.
    """
    def cumsum_in_partition(iter_rows):
        total = 0
        for row in iter_rows:
            total += get_summand(row)
            yield (total, row)
    rdd = rdd.mapPartitions(cumsum_in_partition)
    def last_partition_value(iter_rows):
        final = None
        for cumsum, row in iter_rows:
            final = cumsum
        return (final,)
    partition_sums = rdd.mapPartitions(last_partition_value).collect()
    # partition_cumsums = list(np.cumsum(partition_sums))

    #----from here are the changed lines
    partition_sums = [x for x in partition_sums if x is not None] 
    temp = np.cumsum(partition_sums)
    partition_cumsums = list(temp)
    #----

    partition_cumsums = [0] + partition_cumsums   
    partition_cumsums = sc.broadcast(partition_cumsums)
    def add_sums_of_previous_partitions(idx, iter_rows):
        return ((cumsum + partition_cumsums.value[idx], row)
            for cumsum, row in iter_rows)
    rdd = rdd.mapPartitionsWithIndex(add_sums_of_previous_partitions)
    return rdd

#test on random integer frequency
x = np.random.randint(10, size=1000)
D = sqlCtx.createDataFrame(pd.DataFrame(x.tolist(),columns=['D']))
c = D.groupBy('D').count().orderBy('D')
c_rdd =  c.rdd.map(lambda x:x['count'])
cumsums, values = zip(*cumsum(c_rdd,lambda x: x).collect())

Answer 2

Spark内置了对hive分析/窗口功能的支持，使用分析功能可以轻松实现累积和。

Hive wiki分析/窗口功能。

例子：

假设您有sqlContext对象-

val datardd = sqlContext.sparkContext.parallelize(Seq(("a",1),("b",2), ("c",3),("d",4),("d",5),("d",6)))
import sqlContext.implicits._

//Register as test table
datardd.toDF("id","val").createOrReplaceTempView("test")

//Calculate Cumulative sum
sqlContext.sql("select id,val, " +
  "SUM(val) over (  order by id  rows between unbounded preceding and current row ) cumulative_Sum " +
  "from test").show()

这种方法会导致以下警告。在executor耗尽内存的情况下，相应地调整作业的内存参数，以处理庞大的数据集。

警告WindowExec：没有为窗口操作定义分区！将所有数据移动到单个分区，这可能会导致严重的性能下降

我希望这有帮助。

Answer 3

>

计算每个分区的部分结果：

val partials = rdd.mapPartitionsWithIndex((i, iter) => {
  val (keys, values) = iter.toSeq.unzip
  val sums  = values.scanLeft(0)(_ + _)
  Iterator((keys.zip(sums.tail), sums.last))
})

收集部分和

val partialSums = partials.values.collect

计算分区上的累积和并广播它：

val sumMap = sc.broadcast(
  (0 until rdd.partitions.size)
    .zip(partialSums.scanLeft(0)(_ + _))
    .toMap
)

计算最终结果：

val result = partials.keys.mapPartitionsWithIndex((i, iter) => {
  val offset = sumMap.value(i)
  if (iter.isEmpty) Iterator()
  else iter.next.map{case (k, v) => (k, v + offset)}.toIterator
})