转换kStream窗口聚合的结果

Question

我有1分钟的窗口，用于通过我的一个主题之一。我希望取得该窗口的结果，并再改变一次，以获取我的最终结果，我想将其推入水槽。这些是我的DSL呼叫，可以完成我的总体。

val sessionProcessorStream = builder.stream("collector-prod", Consumed.`with`(Serdes.String, Serdes.String)) //[String,String]
  .filter((_, value) => filterRequest(value))
  .transform(valTransformer,"valTransformState") //[String,String]
  .groupByKey()
  .windowedBy(SessionWindows.`with`(TimeUnit.MINUTES.toMillis(1))) //[Windowed[String],String]
  .aggregate(sessionInitializer,sessionAggregator,sessionMerger,sessionMaterialized) //[Windowed[String],SessionEvent]
  .toStream      
  .transform(sessionTransformer,"sessionTransformState") // [String,Long]

当我尝试在同一窗口中运行此问题并处理两个事件时，我会收到以下错误：

Exception in thread "9-fc11122f-0db3-401b-bdaa-2480eacb8e74-StreamThread-1" org.apache.kafka.streams.errors.ProcessorStateException: task [1_0] Failed to flush state store session-agg-store7
    at org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:245)
    at org.apache.kafka.streams.processor.internals.AbstractTask.flushState(AbstractTask.java:196)
    at org.apache.kafka.streams.processor.internals.StreamTask.flushState(StreamTask.java:327)
    at org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:307)
    at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208)
    at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:302)
    at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:292)
    at org.apache.kafka.streams.processor.internals.AssignedTasks$2.apply(AssignedTasks.java:87)
    at org.apache.kafka.streams.processor.internals.AssignedTasks.applyToRunningTasks(AssignedTasks.java:452)
    at org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:381)
    at org.apache.kafka.streams.processor.internals.TaskManager.commitAll(TaskManager.java:310)
    at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1018)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:835)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:774)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:744)
Caused by: java.lang.NullPointerException
    at com.fasterxml.jackson.core.JsonFactory.createParser(JsonFactory.java:857)
    at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3079)
    at SessionEventDeserializer.deserialize(SessionEventDeserializer.scala:17)
    at SessionEventDeserializer.deserialize(SessionEventDeserializer.scala:8)
    at org.apache.kafka.streams.state.StateSerdes.valueFrom(StateSerdes.java:158)
    at org.apache.kafka.streams.state.internals.CachingSessionStore.putAndMaybeForward(CachingSessionStore.java:173)
    at org.apache.kafka.streams.state.internals.CachingSessionStore.access$000(CachingSessionStore.java:38)
    at org.apache.kafka.streams.state.internals.CachingSessionStore$1.apply(CachingSessionStore.java:88)
    at org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:141)
    at org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:99)
    at org.apache.kafka.streams.state.internals.ThreadCache.flush(ThreadCache.java:127)
    at org.apache.kafka.streams.state.internals.CachingSessionStore.flush(CachingSessionStore.java:196)
    at org.apache.kafka.streams.state.internals.MeteredSessionStore.flush(MeteredSessionStore.java:165)
    at org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:242)
    ... 14 more

我不确定例外的确切原因，但是，我确实认为进入最后一个转换的流是我的键是Windowed[String]。我认为并且希望它是普通的String。

我试图通过在Tostream之后运行.map来解决此问题

  .map { (_,v) => {
      new KeyValue[String,SessionEvent](v.name,v)
  }}  

但是，我得到了我无法弄清楚的编译时间错误。

Error:(106, 8) no type parameters for method map: (x$1: org.apache.kafka.streams.kstream.KeyValueMapper[_ >: org.apache.kafka.streams.kstream.Windowed[String], _ >: SessionEvent, _ <: org.apache.kafka.streams.KeyValue[_ <: KR, _ <: VR]])org.apache.kafka.streams.kstream.KStream[KR,VR] exist so that it can be applied to arguments (org.apache.kafka.streams.kstream.KeyValueMapper[org.apache.kafka.streams.kstream.Windowed[String],SessionEvent,org.apache.kafka.streams.KeyValue[String,SessionEvent]])
 --- because ---
argument expression's type is not compatible with formal parameter type;
 found   : org.apache.kafka.streams.kstream.KeyValueMapper[org.apache.kafka.streams.kstream.Windowed[String],SessionEvent,org.apache.kafka.streams.KeyValue[String,SessionEvent]]
 required: org.apache.kafka.streams.kstream.KeyValueMapper[_ >: org.apache.kafka.streams.kstream.Windowed[String], _ >: SessionEvent, _ <: org.apache.kafka.streams.KeyValue[_ <: ?KR, _ <: ?VR]]
      .map { (_,v) => {
Error:(106, 20) type mismatch;
 found   : org.apache.kafka.streams.kstream.KeyValueMapper[org.apache.kafka.streams.kstream.Windowed[String],SessionEvent,org.apache.kafka.streams.KeyValue[String,SessionEvent]]
 required: org.apache.kafka.streams.kstream.KeyValueMapper[_ >: org.apache.kafka.streams.kstream.Windowed[String], _ >: SessionEvent, _ <: org.apache.kafka.streams.KeyValue[_ <: KR, _ <: VR]]
      .map { (_,v) => {

转换窗户骨料结果的正确方法是什么？我将感谢任何帮助，并很乐意在任何方面进一步澄清。

Answer 1

错误起源于"齐平"，因此我怀疑这是聚集后的变压器，而是聚合本身。我认为您的SessionEventDeserializer中有一个错误，因为它会抛出NullPointerException-您是否正确处理null？您的DeSerializer必须能够处理null并返回null。