基于滑动窗口的流式数据处理示例

　　在科技飞速发展的今天，天天都会产生大量新数据，例如银行交易记录，卫星飞行记录，网页点击信息，用户日志等。为了充分利用这些数据，咱们须要对数据进行分析。在数据分析领域，很重要的一块内容是流式数据分析。流式数据，也即数据是实时到达的，没法一次性得到全部数据。一般状况下咱们须要对其进行分批处理或者以滑动窗口的形式进行处理。分批处理也即每次处理的数据之间没有交集，此时须要考虑的问题是吞吐量和批处理的大小。滑动窗口计算表示处理的数据每次向前移N个单位，N小于要处理数据的长度。例如，在语音识别中，每一个包处理大约25ms的音频数据，而后以步幅10ms向前移动处理下一个包的数据。语音识别就是一个典型的流式数据经过滑动窗口方式进行处理的例子。在本文中，咱们关注N=1的状况，也即每次处理完一个包以后，向前移动一个单位继续处理下一个包，以下图所示。
　　
　　图1 基于滑动窗口的流式数据处理示例
　　
　　咱们主要关注几个常见的数学统计量：最小（大）值、平均值和中位数。事实上，只要知道了最大值和最小值的求法，很容易计算极差；知道了平均值的求法，就能够很容易地计算方差和标准差。针对上述统计量的计算都有一个naïve算法，也即不考虑先后两个包之间数据重叠，将每一个包当作独立的，对每个包分别计算上述统计量。若是总数据长度为n，每一个包的长度为k，则计算上述统计量的复杂度为O(nk)（针对给定数组求中位数的问题，存在复杂度O(k)的算法，实现方法是基于快排进行改进，网上资料不少在此再也不作介绍）。咱们尝试在naïve算法的基础上下降每一个统计量的计算复杂度，下面开始正式的介绍。
　　
　　1. 最小（大）值
　　
　　这是一个经典问题，一般被称为滑动极值问题。问题描述：给定一个长度为n的数列a0,a1,...,an−1和一个整数k，求数列bi=min{ai,ai+1,...,ai+k−1}(i=0,1,...,n−k)。
　　
　　经过使用单调队列能够在O(n)的时间内解决。单调队列维护数列的下标，队列内的元素知足：
　　
　　设单调队列从头部开始的元素值为xi，则xi<xi+1且axi<axi+1。
　　
　　简单来讲单调队列就是下标对应的元素是严格递增的顺序（固然在实际应用过程当中，可能不严格单调，也多是递减的顺序）。
　　
　　考虑以ai结尾的k个元素，求bi−k+1。假定单调递增队列中维护了ai以前的k-1个元素相关的最小值下标，为了求bi−k+1，咱们须要将ai和单调队列中元素进行比较。当队列末尾的元素j知足aj≥ai，则不断取出末尾元素，直到队列为空或者aj<ai。 ai不只会影响bi−k+1的计算，也会影响后续k-1个bi的计算。若是ai是这一段的最小值，则它在单调队列中就不会被删除，进而能够用O(1)的时间求单个bi。
　　
　　当删除单调队列的元素时，须要判断头部元素是否还须要。若是已经脱离计算bi的范围，则能够删除头部元素。求单个bi的值，只须要返回单调队列的头部元素便可。均摊复杂度为O(n)。求最小值的代码以下：
　　
　　#define MAX_N 100000
　　
　　static int a[MAX_N];
　　
　　static int b[MAX_N];
　　
　　static int deque[MAX_N];
　　
　　void range_min(int n,int k)
　　
　　{
　　
　　int s=0,t=0;//单调队列的头和尾指针
　　
　　for (int i=0;i<n;i++)
　　
　　{
　　
　　//在单调队列的末尾加入i
　　
　　while (s<t&&a[deque[t-1]]>=a[i]) t--;//维护严格的单调递增队列
　　
　　deque[t++]=i;
　　
　　if (i-k+1>=0)
　　
　　{
　　
　　b[i-k+1]=a[deque[s]www.lieqibiji.com];
　　
　　}
　　
　　//从单调队列头部删除元素
　　
　　if (deque[s]==i-k+1)

　　
　　求滑动最大值只须要将大于等于号改成小于等于号便可，维护一个单调递减队列。经过使用单调队列，流式数据中极值计算的复杂度能够由O(nk)降为O(n)，当每一个包的长度很大时，算法的优化效果会很是明显。滑动极值问题具备很普遍的应用，但愿你们能知道这个优雅的解法。单调队列还有不少其余应用场景，好比解决《leetcode之Largest Rectangle in Histogram》。此外，在一些动态规划问题中，它也能够用来下降时间复杂度。
　　
　　2. 平均值
　　
　　滑动平均值的计算比较容易优化，咱们须要作的就是维护区间内元素的和，除以区间元素个数k便是区间平均值。当计算下一个区间的平均值时，咱们先将上一个区间的和减掉上一个区间第一个元素的值，而后加上当前区间最后一个元素的值，而后除以k便是当前区间的平均值。求区间平均值的代码以下：
　　
　　#define MAX_N 100000
　　
　　static int www.boayulevip.cn a[MAX_N];
　　
　　static int b[MAX_N];
　　
　　void range_mean(int n,int k)
　　
　　{
　　
　　int sum=0;
　　
　　for (int i=0;i www.yszxylpt.com <n;i++)
　　
　　{
　　
　　sum+=a[i];
　　
　　if(i-k+1>=0)
　　
　　{
　　
　　b[i-k+1]=sum/k;
　　
　　很明显能够看出上述代码的复杂度为O(n)。求方差能够采用相似的思路，在求和的同时也求一个平方和，以后采用方差的平方和公式便可求得方差。
　　
　　3. 中位数
　　
　　中位数是一个很是重要的指标，在不少应用中都会用到，可是相比前两个统计量，中位数的优化要麻烦不少。
　　
　　在介绍基于滑动窗口的中位数计算以前，咱们先看一个相似的问题：也是流式数据求中位数，可是每次都求前面全部数据的中位数。该问题也很经典，出如今剑指offer一书中，具体解法可参考《数据流中的中位数》。简单来讲，就是构造一个最大堆和一个最小堆，最大堆的元素都小于最小堆中的元素，并且最小堆中的元素个数至多比最大堆中的元素个数多1。每次来新元素的时候，根据当前两个堆的元素个数来决定往哪一个堆插入元素，在插入的同时保证上面所说的两个前提。插入复杂度是O(log n)，查询复杂度是O(1)。
　　
　　基于滑动窗口的中位数计算解法和上面的问题相似，也须要构造一个最大堆和最小堆，同时也知足上面的两个条件，区别就在于咱们每次计算完一次中位数以后，都须要从堆中删除一个最老的元素。能够经过和中位数比较来肯定删除哪一个堆中的元素。一般的堆操做通常是插入和删除堆顶元素，在此须要实现一个函数能够删除任意位置的堆元素，同时保证堆的结构不被破坏，这不是一个困难的问题，实现和删除堆顶元素相似。若是数据是以数组形式一次给定，最老的元素能够经过访问原数组得到，若是流式数据一次只给定一个数据，咱们能够经过循环队列保存最近的k个元素来得到最老的元素。代码实现能够参考博客《找滑动窗口的中位数》，在此就不给出详细代码。每来一个数据都须要执行一次插入和删除，复杂度是O(log k)，因此针对流式数据的中位数问题算法复杂度是O(nlogk)，相比朴素算法也有明显地提高。