Kafka Streamer : 사용자 정의 'Serdes'관련 문제

Question

Kafka 스 트리머로 Confluent-3.2.1을 사용하고 있습니다. 내 KGroupedStream<String, MyClass1>을 KTable<Windowed<String>,MsgAggr>에 집계하려고합니다. 집계를 사용하는 동안 나는 TimeWindows.of(TimeUnit.SECONDS.toMillis(5))도 사용하고 있습니다. 나는 사용자 정의 "Serdes"를 집계에 대한 인수로 사용하고 있습니다.

[2017-05-23 18:16:45,648] ERROR stream-thread [StreamThread-1] Streams application error during processing: (org.apache.kafka.streams.processor.internals.StreamThread:249) 
java.lang.ClassCastException: my.kafka.strm.MyClass1 cannot be cast to java.lang.String 
at org.apache.kafka.common.serialization.StringSerializer.serialize(StringSerializer.java:24) 
at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:64) 
at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:82) 
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:202) 
at org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process(KStreamFilter.java:44) 
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:82) 
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:202) 
at org.apache.kafka.streams.kstream.internals.KStreamMap$KStreamMapProcessor.process(KStreamMap.java:43) 
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:82) 
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:202) 
at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:66) 
at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:180) 
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:436) 
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:242) 
Exception in thread "StreamThread-1" java.lang.ClassCastException: my.kafka.strm.MyClass1 cannot be cast to java.lang.String 
at org.apache.kafka.common.serialization.StringSerializer.serialize(StringSerializer.java:24) 
at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:64) 
at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:82) 
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:202) 
at org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process(KStreamFilter.java:44) 
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:82) 
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:202) 
at org.apache.kafka.streams.kstream.internals.KStreamMap$KStreamMapProcessor.process(KStreamMap.java:43) 
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:82) 
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:202) 
at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:66) 
at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:180) 
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:436) 
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:242) 

Answer 1

문제는 message.groupByKey();을 함께 : 사용자의 코드는 내가 오류를받은 코드를 실행하면

이

Map<String, Object> serdeProps = new HashMap<>(); 

final Serializer<MsgAggr> pageViewSerializer = new JsonPOJOSerializer<>(); 
serdeProps.put("JsonPOJOClass", MsgAggr.class); 
pageViewSerializer.configure(serdeProps, false); 

final Deserializer<MsgAggr> pageViewDeserializer = new JsonPOJODeserializer<>(); 
serdeProps.put("JsonPOJOClass", MsgAggr.class); 
pageViewDeserializer.configure(serdeProps, false); 

final Serde<MsgAggr> pageViewSerde = Serdes.serdeFrom(pageViewSerializer, pageViewDeserializer);` 

스트리밍을위한 코드

KGroupedStream<String, MyClass1> msg_grp = message 
      .groupByKey(); 
KTable<Windowed<String>,MsgAggr> msg_win = msg_grp 
      //.reduce(new Reduced(), arg1, arg2); 
      .aggregate(new Init(), 
        new Aggr(), 
        TimeWindows.of(TimeUnit.SECONDS.toMillis(5)), 
        pageViewSerde, 
        "MySample_out"); 

이다,이다 "의 Serdes"를 정의 . 귀하의 사용자 정의 클래스에 대한 문자열 Serde 사용 MyClass1. MyClass1에 대한 사용자 지정 serializer 및 deserializer를 구현하고 오버로드 된 버전 groupByKey - https://kafka.apache.org/0102/javadoc/org/apache/kafka/streams/kstream/KStream.html#groupByKey(org.apache.kafka.common.serialization.Serde,%20org.apache.kafka.common.serialization.Serde)