Apache Spark Streaming中的近似长期滑动平均值

Question

我想使用Apache Spark Streaming计算一周窗口大小的滑动平均值。应及时交付结果，即延迟在亚分钟以下。在我的例子中，存储一周的流数据是不切实际的，尽管这将需要计算准确的结果(如果我错了，请纠正我)。

因此，我的目标是某种近似。我的想法是使用一个1h的滚动窗口来产生一个1h前的平均结果s_1流。然后，我将使用滑动平均值，并在s_1上持续一周来计算1周前的平均结果s_2。除了s_1和s_2之外，我还会有另一个流s_3，它包含超过1小时的滑动平均结果。然后，我将连接流s_2和s_3，对于连接元组(t_2, t_3)，我将发出(t_2 + t_3) / 2。我还会将包含的元组的最小和最大时间戳附加到每个平均元组。我将使用这些时间戳来防止s_2和s_3元组的重叠。例如：

s_2 tumbling window size 2 (tuples)
s_3 sliding window size 2, interval 1 (tuples)
stream 3  4  9  8  7     

time    s_2    2_3    out
1        -      3      3
2        -     3.5    3.5
3       3.5    6.5    3.5   the s_3 tuple 6.5 is ignored because  min_timestamp(6.5) <= max_timestamp(3.5)
4       3.5    8.5     6   (compute (3.5 + 8.5) / 2
5        6     7.5     6    the s_3 tuple 7.5 is ignored because  min_timestamp(7.5) <= max_timestamp(6)

虽然我能够让它在Apache Storm中工作，但我不知道如何在Spark中做到这一点，因为由于不同的滑动间隔，在Spark中不允许加入s_2和s_3。

所以第一个问题是:我如何在Spark Streaming中实现它？

问题#2:你能想出更好的方法在流处理系统中高效地计算长时间的滑动平均值吗？

Answer 1

有许多策略可以在流上生成相当高精度的近似答案。我们使用的一种策略是在spark中的无限流上使用stratified sampling。我们在一个名为SnappyData的开源项目中将分层采样引入了Apache Spark。嗯，在Spark缺少的其他几个东西中。SnappyData可以在流上维护均匀的随机样本，但通过允许开发人员选择流中重要的列/维度来确保高精度。因此，例如，在您的示例中，确保每分钟或每小时捕获足够的样本。样本以Spark Dataframe/Column表的形式可见，并且可查询。当执行像avg/sum/count/etc这样的聚合查询时，它使用一系列算法来计算答案，而只需要很少的资源和时间。

下面是Snappydata中的伪代码。

Create sample table MyInfiniteStream on <Stream> options (qcs 'min(timestamp), fraction '0.01')
    // Of course, you can use the Dataframe api to do this instead of SQL too. 
    // your DStream <Stream> is registered with SnappyData
    // min(timestamp) tells which columns to use for stratification
    // fraction indicates what percentage of the input data to retain in the sample. 

然后，您可以直接在此上运行Spark SQL查询，无论是否有错误约束。现在的好处是，您的时间间隔可以是非常细粒度的，也可以是粗略的。

select avg(myMeasureColumn), dimension d from MyInfiniteStream group by d with Error 0.1
// this would ensure the result is always at least 90% accurate. 
select avg(myMeasureColumn), dimension d from MyInfiniteStream where timestamp >x and timestamp < y group by d with Error 0.1

你可以对ideas here有更好的理解。SnappyData与Spark完全兼容。

您也可以尝试直接在Spark中实现，尽管没有提供在线采样(即直接在流上)，也没有内置算法来计算具有置信区间的误差。签出Dataset上的“sample”方法。