删除KStream-KStream连接在Kafka Streams中的重复中间结果

Question

我有以下情况：

1. 表A和表B使用FK连接

如何丢弃[a， null]？

一个选项是执行innerJoin，但在update查询的情况下，这仍然是一个问题。

我们尝试使用事件时间戳进行过滤（即使用最新的时间戳保留事件），但时间戳的唯一性无法保证。

最终目标是能够识别最新的聚合，以便我们可以在查询时过滤出中间结果（在Athena/Presto或某些RDBMS中）。

Answer 1

目前，我发现最好的工作方法是利用输出记录中的Kafka偏移量。

该方法可概括为：

1. 执行您想要执行的所有逻辑，而不必担心同一密钥的多个记录

现在，您的输出主题包含同一个键的多条消息，但每条消息的偏移量不同。

在查询期间，现在可以使用子查询为每个键选择最大偏移量。

下面可以看到一个变压器供应商示例

/**
 * @param <K> key type
 * @param <V> value type
 */
public class OutputTransformSupplier<K, V> implements TransformerSupplier<K, V, KeyValue<String, String>> {
  @Override
  public Transformer<K, V, KeyValue<String, String>> get() {
    return new OutputTransformer<>();
  }

  private class OutputTransformer<K, V> implements Transformer<K, V, KeyValue<String, String>> {
    private ProcessorContext context;

    @Override
    public void init(ProcessorContext context) {
      this.context = context;
    }

    /**
     * @param key   the key for the record
     * @param value the value for the record
     */
    @Override
    public KeyValue<String, String> transform(K key, V value) {
      if (value != null) {
        value.setKafkaOffset(context.offset());
      }
      return new KeyValue<>(key, value);
    }

    @Override
    public KeyValue<String, String> punctuate(long timestamp) {
      return null;
    }

    @Override
    public void close() {
      // nothing to close
    }
  }
}