将输入主题的每个传入记录与其单个precede记录进行比较

Question

事实上，直到现在我还没有成功。您能帮助我，请提供我的详细代码示例，以实现使用Kafka流DSL？

Answer 1

您可以使用处理器API。

您必须实现Transformer接口方法转换：

>

将查找键值，

object SampleApp extends App {

  val storeName: String = ???
  val builder: StreamsBuilder = new StreamsBuilder()
  builder.stream("topicName")(Consumed.`with`(Serdes.String(), Serdes.String()))
    .transform[String, String](() => SampleTransformer(storeName))
    .to("outputTopic")(Produced.`with`(Serdes.String(), Serdes.String()))
}


case class SampleTransformer(storeName: String)
  extends Transformer[String, String, KeyValue[String, String]]
    with LazyLogging {

  var store: KeyValueStore[String, String] = _

  override def init(context: ProcessorContext): Unit = {
    super.init(context)
    store = context.getStateStore(storeName).asInstanceOf[KeyValueStore[String, String]]
  }

  override def transform(key: String, newValue: String): String = {
    val valueToPass = Option(store.get(key)).map(oldValue => someComputation(oldValue, newValue))
    store.put(key, newValue)
    valueToPass.orNull
  }

  def someComputation(oldValue: String, newValue: String): String = ???

  override def close(): Unit = {
    // Close
  }
}