SparkStreaming 对Window的reduce的方法解析（转载）

在sparkstreaming中对窗口范围进行reduce主要有下面四个方法，其他方法都是直接或者间接调用下面的方法来获取结果

对非(K,V)形式的RDD 窗口化reduce：

1.reduceByWindow(reduceFunc, windowDuration, slideDuration) 
2.reduceByWindow(reduceFunc, invReduceFunc, windowDuration, slideDuration)

对(K,V)形式RDD 按Key窗口化reduce:

1.reduceByKeyAndWindow(reduceFunc, windowDuration, slideDuration) 
2.reduceByKeyAndWindow(reduceFunc, invReduceFunc, windowDuration, slideDuration, numPartitions, filterFunc)

从方法上面来看，理解和使用#1的方法确实非常简单，但是在大数据计算过程中，我们还是应该把效率放到第一位，如果简单的看一下源码，或许可以明白为什么我们应该在允许的情况下尽量用#2的方法。

reduceByWindow(reduceFunc, windowDuration, slideDuration) 代码：

可以看到他做了两次reduce，第一次对整个self做一次reduce，然后截取时间区间，对结果再做一次reduce。
第一点：对整个self做reduce会比较慢，因为self都是相对比较大的集合。
第二点：进行了两次reduce

def reduceByWindow(  
    reduceFunc: (T, T) => T,  
    windowDuration: Duration,  
    slideDuration: Duration  
  ): DStream[T] = ssc.withScope {  
  this.reduce(reduceFunc).window(windowDuration, slideDuration).reduce(reduceFunc)  
}

如果我们看：
reduceByWindow(reduceFunc, invReduceFunc, windowDuration, slideDuration)

实际上他是调用了效率非常高的reduceByKeyAndWindow(reduceFunc, invReduceFunc, windowDuration, slideDuration, numPartitions, filterFunc) 方法 ==》
这样的话其实他只对newRDDs和oldRDDs做reduce，由于这两个RDDs都非常小，可以想象效率是非常高的

def reduceByWindow(  
    reduceFunc: (T, T) => T,  
    invReduceFunc: (T, T) => T,  
    windowDuration: Duration,  
    slideDuration: Duration  
  ): DStream[T] = ssc.withScope {  
    this.map(x => (1, x))  
        .reduceByKeyAndWindow(reduceFunc, invReduceFunc, windowDuration, slideDuration, 1)  
        .map(_._2)  
}

如果看reduceByKeyAndWindow的话，情况也是一样，一个是执行：

self.reduceByKey(reduceFunc, partitioner)  
        .window(windowDuration, slideDuration)  
        .reduceByKey(reduceFunc, partitioner)

而另外一个确是在已有的window值基础上做了简单的加加减减

例如：

1、reduceByKeyAndWindow（_+_,Seconds(3), Seconds(2)）

可以看到我们定义的window窗口大小Seconds(3s) ，是指每2s滑动时，需要统计前3s内所有的数据。

2、对于他的重载函数reduceByKeyAndWindow（_+_,_-_,Seconds(3s),seconds(2)）

设计理念是，当滑动窗口的时间Seconds(2) < Seconds(3)（窗口大小）时，两个统计的部分会有重复，那么我们就可以

不用重新获取或者计算，而是通过获取旧信息来更新新的信息，这样即节省了空间又节省了内容，并且效率也大幅提升。

如上图所示，2次统计重复的部分为time3对用的时间片内的数据，这样对于window1，和window2的计算可以如下所示

win1 = time1 + time2 + time3

win2 = time3 + time4 + time5

更新为

win1 = time1 + time2 + time3

win2 = win1+ time4 + time5 - time2 - time3

这样就理解了吧, _+_是对新产生的时间分片（time4,time5内RDD）进行统计，而_-_是对上一个窗口中，过时的时间分片

(time1,time2) 进行统计

宗上，从效率上面考虑，我们应该尽量使用包含invReduceFunc的方法，同样情况下摒弃只有reduceFunc的方法

参考文献：

http://humingminghz.iteye.com/blog/2308231

http://www.cnblogs.com/zDanica/p/5471592.html

https://yq.aliyun.com/articles/60315