Kafka Streams如何在scala中从kafka消息中获取TimeStamp

Question

我正在运行一个简单的Kafka streams应用程序，它将使用Node JS记录的信息带到一个Kafka主题。

E.g. 
  Producer = kafka.Producer
  KeyedMessage = kafka.KeyedMessage
  client = new kafka.KafkaClient()
  producer = new Producer(client)
  km = new KeyedMessage('key', 'message')
  kafka_message = JSON.stringify({ id: req.session.data.toString(), url: article.info })
  payloads = [
    { topic: 'eventTopic', messages: kafka_message,timestamp:timestampNow}
  ];
  producer.send(payloads, function (err, data) {
    console.log(data);
  });

还需要注意的是，时间戳只是一个数字，表示自1970年6月以来的秒数。

我使用scala中的Kafka流来使用这些数据。

例如。

val builder = new StreamsBuilder

val stream = builder
    .stream[String, String]("TopicTest")
    .foreach((k:String, v:String) => {
     println(k)
     println(v) 
}

然而，我不确定如何将时间戳（我从nodeJS发送的）提取到这个流中。

例如，如果我尝试做这样的事情

val stream = builder
    .stream[String, String,Long]("TopicTest")
    .foreach((k:String, v:String,timeStamp:Long) => {
     println(k)
     println(v) 
     println(timeStamp)
}

这会导致错误“无法解析符号流”。我在想我该怎么解决这个问题。这里是我的拓扑结构和流的配置，仅供参考。val topology=builder。建筑

  import java.util.Properties
  val props = new Properties()
  import org.apache.kafka.streams.StreamsConfig

  val appId = this.getClass.getSimpleName.replace("$", "")
  props.put(StreamsConfig.APPLICATION_ID_CONFIG, appId)
  props.put(StreamsConfig.CLIENT_ID_CONFIG, appId)
  props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, ":9092")
  props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")


  // Step 4. Create Kafka Streams Client
  import org.apache.kafka.streams.KafkaStreams
  val ks = new KafkaStreams(topology, props)


  ks.start

Answer 1

确实存在一个TimeStampExtractor（https://jaceklaskowski.gitbooks.io/mastering-kafka-streams/content/kafka-streams-TimestampExtractor.html）。但是，可以将时间戳作为任何常规kafka消息发送到。我更改的第一件事是我的NodeJS代码。

kafka_message = JSON.stringify({ id: req.session.information.toString(), url: article.info,timestamp:timestampNow.toString() })
  payloads = [
    { topic: 'eventTopic', messages: kafka_message}
  ];
 
  producer.send(payloads, function (err, data) {
    console.log(data);
  });

我发送的JSON消息中现在有一个时间戳字段。

最后，我们可以使用argonaut解析JSON消息。

 val streamEvents = builder
    .stream[String, String]("testTopic")
    .foreach((k:String, json:String) => {
      println(k)
      println(json)
      println(Parse.parse(json))
      val url:String = Parse.parseWith(json, _.field("url").flatMap(_.string).getOrElse("Error!"), msg => msg)
      val id:String = Parse.parseWith(json, _.field("id").flatMap(_.string).getOrElse("Error!"), msg => msg)
      val timestamp:String = Parse.parseWith(json, _.field("timestamp").flatMap(_.string).getOrElse("Error!"), msg => msg)
      val timeStampInt:Long = timestamp.toLong