使用 Kafka Streams DSL 进行两步窗口聚合

假设我有一个流“stream-1”，每秒由 1 个数据点组成，我想计算一个派生流“stream-5”，其中包含使用 5 秒的跳跃窗口和另一个流“stream-10”的总和它基于包含使用 10 秒跳跃窗口的总和的“stream-5”。需要分别对每个键进行聚合，我希望能够在不同的进程中运行每个步骤。如果stream-5和stream-10包含相同密钥/时间戳的更新，这本身不是问题（所以我不一定需要如何发送时间窗口 KTable 的最终 kafka-streams 聚合结果？）只要最后的值是正确的。

有没有一种（简单的）方法可以使用高级 Kafka Streams DSL 来解决这个问题？到目前为止，我还没有找到一种优雅的方法来处理由于聚合而在stream-5 上产生的中间更新。

我知道中间更新可以通过某种方式控制cache.max.bytes.buffering and commit.interval.ms设置，但我认为任何设置都不能保证在所有情况下都不会产生中间值。另外，我可以尝试使用密钥的时间戳部分在读取时将“stream-5”转换为 KTable，但似乎 KTable 不支持像 KStreams 那样的窗口操作。

这是我到目前为止所拥有的，由于 Stream-5 上的中间聚合值而失败

Reducer<DataPoint> sum = new Reducer<DataPoint>() {                                                                           
    @Override                                                                                                                 
    public DataPoint apply(DataPoint x, DataPoint y) {                                                                        
        return new DataPoint(x.timestamp, x.value + y.value);                                                                 
    }                                                                                                                         
 };                                                                                                                           

 KeyValueMapper<Windowed<String>, DataPoint, String> strip = new 
           KeyValueMapper<Windowed<String>, DataPoint, String>() {      
      @Override                                                                                                               
      public String apply(Windowed<String> wKey, DataPoint arg1) {                                                            
          return wKey.key();                                                                                                  
      }                                                                                                                       
 };                                                                                                                           

KStream<String, DataPoint> s1 = builder.stream("stream-1");                                                                      

s1.groupByKey()                                                                                                               
       .reduce(sum, TimeWindows.of(5000).advanceBy(5000))                                                                     
       .toStream()                                                                                                            
       .selectKey(strip)                                                                                                      
       .to("stream-5");                                                                                                          

KStream<String, DataPoint> s5 = builder.stream("stream-5");                                                                      

s5.groupByKey()                                                                                                               
       .reduce(sum, TimeWindows.of(10000).advanceBy(10000))                                                                   
       .toStream()                                                                                                            
       .selectKey(strip)                                                                                                      
       .to("stream-10");      

现在如果stream-1包含输入（键只是KEY）

KEY {"timestamp":0,"value":1.0}
KEY {"timestamp":1000,"value":1.0}
KEY {"timestamp":2000,"value":1.0}
KEY {"timestamp":3000,"value":1.0}
KEY {"timestamp":4000,"value":1.0}
KEY {"timestamp":5000,"value":1.0}
KEY {"timestamp":6000,"value":1.0}
KEY {"timestamp":7000,"value":1.0}
KEY {"timestamp":8000,"value":1.0}
KEY {"timestamp":9000,"value":1.0}

Stream-5 包含正确的（最终）值：

KEY {"timestamp":0,"value":1.0}
KEY {"timestamp":0,"value":2.0}
KEY {"timestamp":0,"value":3.0}
KEY {"timestamp":0,"value":4.0}
KEY {"timestamp":0,"value":5.0}
KEY {"timestamp":5000,"value":1.0}
KEY {"timestamp":5000,"value":2.0}
KEY {"timestamp":5000,"value":3.0}
KEY {"timestamp":5000,"value":4.0}
KEY {"timestamp":5000,"value":5.0}

但stream-10是错误的（最终值应该是10.0），因为它还考虑了stream-5上的中间值：

KEY {"timestamp":0,"value":1.0}
KEY {"timestamp":0,"value":3.0}
KEY {"timestamp":0,"value":6.0}
KEY {"timestamp":0,"value":10.0}
KEY {"timestamp":0,"value":15.0}
KEY {"timestamp":0,"value":21.0}
KEY {"timestamp":0,"value":28.0}
KEY {"timestamp":0,"value":36.0}
KEY {"timestamp":0,"value":45.0}
KEY {"timestamp":0,"value":55.0}

问题在于，所有聚合的结果都是 KTable，这意味着为其输出主题生成的记录代表了一个变更日志。但是，当您随后将它们作为流加载时，下游聚合将重复计数。

相反，您需要将中间主题加载为表，而不是流。但是，您将无法对它们使用窗口聚合，因为它们仅在流上可用。

您可以使用以下模式来完成表而不是流的窗口聚合：

https://cwiki.apache.org/confluence/display/KAFKA/Windowed+aggregations+over+successively+increasing+timed+windows

如果您想在单独的进程中运行每个步骤，您可以调整它，只需记住使用 builder.table() 而不是 builder.stream() 加载中间表。