KStream批处理窗口

3 回答

• 3

  现在（截至Kafka 0.10.0.0 / 0.10.0.1）：您描述的窗口行为是“按预期工作” . 也就是说，如果您收到1,000条传入消息，您（目前）将始终看到最新版本的Kafka / Kafka Streams向下游发送1,000个更新 .

  展望未来：Kafka社区正在开发新功能，以使此更新速率行为更加灵活（例如，允许您在上面描述的所需行为） . 有关详细信息，请参阅KIP-63: Unify store and downstream caching in streams .

  回复于 2024-05-02T10:04:06+08:00
• 4

  我的实际任务是将更新从流推送到redis，但即使redis速度很快，我也不想单独读取/更新/写入 . 我现在的解决方案是使用KStream.process（）提供一个处理器，该处理器在进程中添加队列并实际处理队列中的队列 .

  public class BatchedProcessor extends AbstractProcessor{
  
  ...
  BatchedProcessor(Writer writer, long schedulePeriodic)
  
  @Override
  public void init(ProcessorContext context) {
      super.init(context);
      context.schedule(schedulePeriodic);
  }
  
  @Override
  public void punctuate(long timestamp) {
      super.punctuate(timestamp);
      writer.processQueue();
      context().commit();
  }
  
  @Override
  public void process(Long aLong, IntentUpdateEvent intentUpdateEvent) {
      writer.addToQueue(intentUpdateEvent);
  }
  

  我仍然需要测试，但它解决了我遇到的问题 . 人们可以很容易地以非常通用的方式编写这样的处理器 . API非常整洁干净但是processBatched（（List batchedMessaages） - > ...，timeInterval或countInterval）只是使用punctuate来处理批处理并在那时提交并在Store中收集KeyValues可能是一个有用的补充 .

  但也许它的目的是通过一个处理器解决这个问题，并将API纯粹保留在一个消息中，同时保持低延迟焦点 .

  回复于 2024-05-02T10:04:06+08:00
• 5

  ====== Update ======

  在进一步测试时，这不起作用 . 正确的方法是使用@friedrich-nietzsche概述的处理器 . 我投票给自己的答案.... grrrr .

  ===================

  我仍在努力使用这个API（但是我很喜欢它，所以现在花了很多时间:)），我不确定你要从代码示例结束的地方下游完成什么，但它看起来与我得到的相似工作 . 高水平是：

  从源读取对象 . 它代表一个密钥和1：∞事件数，我想每5秒发布一个事件的总事件数（或TP5s，每5秒事务数） . 代码的开头看起来一样，但我使用：

  • KStreamBuilder . stream

  • reduceByKey

  • 到一个窗口（5000）

  • 到new stream，每5秒获取每个键的累计值 .

  • map，每个键流到一个新的KeyValue

  • to接收器主题 .

  在我的情况下，每个窗口期间，我可以将每个键的所有事件减少到一个事件，这样就行了 . 如果你想保留每个窗口的所有单独事件，我假设可以使用reduce将每个实例映射到一个实例集合（可能使用相同的密钥，或者你可能需要一个新密钥）并在每个窗口期结束时，下游流将获得一堆你的事件集合（或者可能只是所有事件的一个集合），一气呵成 . 它看起来像这样，消毒和Java 7-ish：

  builder.stream(STRING_SERDE, EVENT_SERDE, SOURCE_TOPICS)
          .reduceByKey(eventReducer, TimeWindows.of("EventMeterAccumulator", 5000), STRING_SERDE, EVENT_SERDE)            
          .toStream()
          .map(new KeyValueMapper<Windowed<String>, Event, KeyValue<String,Event>>() {
              public KeyValue<String, Event> apply(final Windowed<String> key, final Event finalEvent) {
                  return new KeyValue<String, Event>(key.key(), new Event(key.window().end(), finalEvent.getCount());
              }
      }).to(STRING_SERDE, EVENT_SERDE, SINK_TOPIC);
  

  回复于 2024-05-02T10:04:06+08:00