compare each individual incoming record from input topic with its individual precede record

Question

I am new in Kafka stream, my use case is to compare the value of each individual incoming record from input topic with value from its individual precede record, and, if a comparison condition is true then send new record containing a comparison result with an index of each compared records into result topic, else do not send anything. (note that all incoming records may have a unique Key for each record or null key).

Doing this in Kafka consumer and producer API is very easy, but ((without using external DB to store the preceding record)) I try to use (Kafka streams DSL API) only, (which include KTable and KStream, with them internal methods such as, aggregate, reduce .. etc) but maybe because I am a beginner so I did not find a clear way to access (internal state store) in these API in order to store state and retrieve the previous record state in order to compare it with current one, then after store the current record instead of old one in order to compare it with the next incoming record. A few approaches try to use Processor API instead of Stream DSL API, but it includes much complexity and I did not fully understand it. This why I try to solve my problem with Stream DSL API. But till now I did not succeed, unfortunately.

Actually, until now I did not succeed. can you help me, please by providing me with detailed code example to get this done using Kafka Stream DSL?

Please edit your question to include what you've tried and what specifically isn't working — OneCricketeer
@cricket_007, I edited my question to meet what you ask. I hope that — Sarru Uken

Bartosz Wardziński Bartosz Wardziński · Accepted Answer · 2018-12-23T19:39:02

You can use Processor API.

You have to implement Transformer interface method transform that:

Will lookup value for key,

1.1 if not present, put value in store and finished

1.2 if present, calculate value, save new value in store and passed result to output topic

Sample code:

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
  }
}

compare each individual incoming record from input topic with its individual precede record

1 Answers