Kafka流加窗口的关键是人类可读

Question

我正在对kafka流进行窗口聚合。它工作正常并进行正确的聚合。这是scala中的代码。CallRecord是一个案例类。

    builder
  .stream[String, String](input_topic)
  .mapValues((elm) => {
    parse(elm).extract[CallRecord]
  })
  .groupBy((key, value) => {
    value.agentId
  })
  .windowedBy(every15Minute)
  .aggregate(CallRecordAggByAgent.empty)((_, callRecord, aggregator) => {
    CallRecordAggByAgent(
      callRecord.agentId,
      ((callRecord.durationMinutes + aggregator.durationMinutesAvg) / aggregator.count).roundD,
      ((callRecord.waitingMinutes + aggregator.waitingMinutesAvg) / aggregator.count).roundD,
      ((callRecord.customerScore + aggregator.scoreAvg) / aggregator.count).roundD,
      aggregator.count + 1
    )
  })
  .mapValues((elm) => {
    write(elm)
  })
  .toStream
  .to(output_topic)

当我试图从KSQLDB中读取这个内容时，当我创建一个关于这个主题的流时，我看到rowkey的值如下3w�H�@ 我知道这是反序列化问题，但我希望能够在KSQL中直接反序列化，或者在流式传输到输出主题时使其长度达到毫秒。我的理解是这应该很容易实现，但我想我错过了一些细微差别。




Answer 1

我给出的解决方案如下。显然这并不难。

import io.circe.generic.auto._
import org.json4s._
import org.json4s.native.Serialization.write

builder
      .stream[String, String](args.INPUT_TOPIC)
      .mapValues((elm) => {
        parse(elm).extract[CallRecord]
      })
      .groupBy((key, value) => {
        value.agentId
      })
      .windowedBy(every15Minute)
      .aggregate(CallRecordAggByAgent.empty)((_, callRecord, aggregator) => {
        CallRecordAggByAgent(
          callRecord.agentId,
          ((callRecord.durationMinutes + aggregator.durationMinutesAvg) / aggregator.count).roundD,
          ((callRecord.waitingMinutes + aggregator.waitingMinutesAvg) / aggregator.count).roundD,
          ((callRecord.customerScore + aggregator.scoreAvg) / aggregator.count).roundD,
          aggregator.count + 1
        )
      })
      .mapValues((elm) => {
        write(elm)
      })
      .toStream
      .selectKey((k, v) => {
        s"${k.key()}@${k.window().startTime().toEpochMilli.toString}"
      })
      .to(args.OUTPUT_TOPIC)

selectKey提供了更改分组键的可能性，因此在流式传输到输出主题之前，我从键中提取时间戳，并将其设置为字符串。