如何在Kafka流中添加冷却/速率限制？

Question

我所尝试的：

使用aggreate作为有状态转换，但聚合状态依赖于时间。然而，即使得到的KTable在聚合值中没有变化，但是KTable（作为changelog）将继续向下发送元素，因此无法实现对流进行“速率限制”的预期效果

val fooStream: KStream[String, String] = builder.stream("foobar2")
fooStream
  .groupBy((key, string) => string)
  .aggregate(() => "constant",
    (aggKey: String, value: String, aggregate: String) => aggregate,
    stringSerde,
    "name")
  .print

提供以下输出：

[KSTREAM-AGGREGATE-0000000004]: string , (constant<-null)
[KSTREAM-AGGREGATE-0000000004]: string , (constant<-null)

可能的重复涉及到记录缓存如何在聚合决定在下游发布元素时引起一些混乱的问题。然而，主要的问题是如何在DSL中实现“速率限制”。正如@Miguno所指出的那样，必须返回到较低级别的处理器API。下面我贴了一个相当冗长的方法：

  val logConfig = new util.HashMap[String, String]();
  // override min.insync.replicas
  logConfig.put("min.insyc.replicas", "1")

  case class StateRecord(alert: Alert, time: Long)

  val countStore = Stores.create("Limiter")
    .withKeys(integerSerde)
    .withValues(new JsonSerde[StateRecord])
    .persistent()
    .enableLogging(logConfig)
    .build();
  builder.addStateStore(countStore)

  class RateLimiter extends Transformer[Integer, Alert, KeyValue[Integer, Alert]] {
    var context: ProcessorContext = null;
    var store: KeyValueStore[Integer, StateRecord] = null;

    override def init(context: ProcessorContext) = {
      this.context = context
      this.store = context.getStateStore("Limiter").asInstanceOf[KeyValueStore[Integer, StateRecord]]
    }

    override def transform(key: Integer, value: Alert) = {
      val current = System.currentTimeMillis()
      val newRecord = StateRecord(value._1, value._2, current)
      store.get(key) match {
        case StateRecord(_, time) if time + 15.seconds.toMillis < current => {
          store.put(key, newRecord)
          (key, value)
        }
        case StateRecord(_, _) => null
        case null => {
          store.put(key, newRecord)
          (key, value)
        }
      }
    }
  }

Answer 1

假设我有一个(User，Alert)元组流。我想要的是对每个用户的流进行速率限制。即。我想要一个只为用户输出一次警报的流。在接下来的60分钟内，用户的任何传入警报都应该被吞没。在这60分钟之后，一个传入警报将再次触发。

当使用Kafka流的DSL时，这目前是不可能的。相反，您可以（并且需要）使用较低级别的处理器API手动实现这样的行为。

注意：我们一直在Kafka社区讨论是否要在DSL中添加这样的功能（通常称为“触发器”）。到目前为止，决定是暂时不具备这种功能。