使用Kafka流在输出中设置时间戳

Question

我在一个Kafka主题“原始数据”中获取CSV，目标是通过在另一个主题“数据”中发送具有正确时间戳（每行不同）的每行来转换它们。

null

我想通过直接设置时间戳来删除这个“内部”主题的使用，但我找不到一个方法（时间戳提取器只在消耗时间使用）。

我在文档中偶然发现了这一行：

请注意，通过调用#forward()时显式地为输出记录分配时间戳，可以在处理器API中更改description默认行为。

2018-01-01,hello 2018-01-02,world（这是一条消息，而不是两条消息）

我想在另一个主题中获得两条消息，其中Kafka记录的时间戳设置为他们的事件时间（2018-01-01和2018-01-02)，而不需要中间主题。

Answer 1

为输出设置时间戳需要Kafka Streams2.0并且仅在处理器API中支持。如果使用DSL，则可以使用transform()来使用这些API。

正如您指出的，您将使用context.forward()。这一呼吁将是：

stream.transform(new TransformerSupplier() {
  public Transformer get() {
    return new Transformer() {
      // omit other methods for brevity
      // you need to get the `context` from `init()`

      public KeyValue transform(K key, V value) {
        // some business logic

        // you can call #forward() as often as you want
        context.forward(newKey, newValue, To.all().withTimestamp(newTimestamp));

        return null; // only return data via context#forward()
      }
    }
  }
});