计算给定窗口的流的统计信息

Question

我有一个经常打勾的计时器KStream（想想秒数），我想在24小时的窗口内计算各种统计数据。例如，24小时更改，即给定点与之前24小时之间的价格差异。

我输出的输出是：

t1 -> t1c1
t2 -> t1c2
t3 -> t1c3

其中t1是输入股票代码，t1c1是输入股票代码，其前面的24小时窗口计算了额外的统计数据。

我已经考虑过这样做的一些方法没有奏效：*我的自动收报机流量大小为24小时，1秒钟跳跃。

builder.stream(rawPriceTickerTopic, ...)
            .groupByKey()
            .windowedBy(
                    TimeWindows.of(TimeUnit.DAYS.toMillis(1))
                    .advanceBy(TimeUnit.SECONDS.toMillis(1))
            .reduce((value1, value2) ->
                    value1.tickerWithStatsFrom(value2), ...)                
            .toStream();

但是，这会产生大量的输出点，因为每个输入报价器为其所属的每个窗口生成输出报价器。

• 保持某种时间序列存储是最新的，从商店24小时前获取值，并从中计算我的统计信息，但这似乎与流的点相反。

Answer 1

我在这里的最终解决方案是放弃窗口并简单地聚合我的代码，在聚合器中维护我自己的24小时窗口。这仍然不是最好的方式，并且有一种唠叨的感觉，我可以用Kafka内置的窗口概念解决它。

如上所述，我使用聚合器进行简单聚合：

streamBuilder.stream(tickerTopic, Consumed.with(...)
                .groupByKey()
                .aggregate(MyAggregator::new,
                        (key, value, aggregate) -> aggregate.addTicker(value),
                        Materialized.with(...)
                .toStream()

结果是，对于原始自动收报器流中的每个记录，我在输出流中得到一个聚合值。我的聚合器逻辑很简单：

• 在订购的集合中添加新的自动收报机。
• 丢弃这个新的最新报价单超过24小时的任何代码。
• 计算新的24小时更改。

（此技术可用于给定窗口上的任何类型的计算，例如移动平均值。）

聚合器的示例代码：

public class MyAggregator {

    private BigDecimal change;

    private TreeSet<Ticker> orderedTickers = new TreeSet<>(MyAggregator::tickerTimeComparator);

    public MyAggregator () {
        this.windowMilis = 86400000;
    }

    public MyAggregator addTicker(Ticker ticker) {
        orderedTickers.add(ticker);
        cleanOldTickers();
        change = getLatest().getAsk().subtract(getEarliest().getAsk());
        return this;
    }

    public BigDecimal getChange() {
        return change;
    }

    public Ticker getEarliest() {
        return orderedTickers.first();
    }

    public Ticker getLatest() {
        return orderedTickers.last();
    }

    private void cleanOldTickers() {
        Date endOfWindow = latestWindow();

        Iterator<Ticker> iterator = orderedTickers.iterator();
        while(iterator.hasNext()) {
            Ticker next = iterator.next();
            if (next.getTimestamp().before(endOfWindow)) {
                iterator.remove();
            }
            // The collection is sorted by time so if we get here we can break.
            break;
        }
    }

    private Date latestWindow() {
        return new Date(getLatest().getTimestamp().getTime() - windowMilis);
    }

    private static int tickerTimeComparator(Ticker t1, Ticker t2) {
        return t1.getTimestamp().compareTo(t2.getTimestamp());
    }

}