You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by Debasish Ghosh <gh...@gmail.com> on 2017/06/20 11:08:10 UTC
facing serialization issues with Kafka Streams application ..
Hi -
I have the following Scala snippet in a Kafka streams application ..
val builder = new KStreamBuilder()
val logRecords: KStream[Array[Byte], LogRecord] =
builder.stream(Serdes.ByteArray(), logRecordSerde, config.toTopic)
val m: KStream[Array[Byte], String] = logRecords.mapValues(hostExtractor)
val n: KStream[String, String] = m.map(
new KeyValueMapper[Array[Byte], String, KeyValue[String, String]]() {
override def apply(key: Array[Byte], value: String): KeyValue[String,
String] = new KeyValue(value, value)
})
val counts: KTable[String, java.lang.Long] =
n.groupByKey().count("HostAggregateCount")
In the above snippet logRecordSerde is a custom implementation of a Serde
based on a Scala case class.
When I run this application, I get the following exception ..
Exception in thread "StreamThread-1"
> org.apache.kafka.streams.errors.StreamsException: Exception caught in
> process. taskId=0_0, processor=KSTREAM-SOURCE-0000000000,
> topic=processed-log, partition=0, offset=0
> at
> org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:216)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:627)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:361)
> Caused by: org.apache.kafka.streams.errors.StreamsException: *A
> serializer (key: org.apache.kafka.common.serialization.ByteArraySerializer
> / value: org.apache.kafka.common.serialization.ByteArraySerializer) is not
> compatible to the actual key or value type (key type: java.lang.String /
> value type: java.lang.String).* Change the default Serdes in StreamConfig
> or provide correct Serdes via method parameters.
> at
> org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:81)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:83)
Possibly the exception comes when I try to do a map on the KStream, which
is supposed to return KStream[String, String]. My question is how do I
inject the proper Serde here ? map does not take any Serde as arguments.
The initial topic from where I read has ByteArray and a custom Serde and it
works fine. But then the map blows up ..
Any help will be appreciated ..
regards.
--
Debasish Ghosh
http://manning.com/ghosh2
http://manning.com/ghosh
Twttr: @debasishg
Blog: http://debasishg.blogspot.com
Code: http://github.com/debasishg
Re: facing serialization issues with Kafka Streams application ..
Posted by Guozhang Wang <wa...@gmail.com>.
Damian,
This type of questions seem to be common, could you add an entry into FAQ?
https://cwiki.apache.org/confluence/display/KAFKA/FAQ
Guozhang
On Tue, Jun 20, 2017 at 8:23 AM, Debasish Ghosh <gh...@gmail.com>
wrote:
> On Tue, Jun 20, 2017 at 7:03 PM, Damian Guy <da...@gmail.com> wrote:
>
> > m.groupBy(mapper, Serdes.String(), Serdes.String()).count("
> > HostAggregateCount")
> >
>
> Thanks! It works ..
>
>
> --
> Debasish Ghosh
> http://manning.com/ghosh2
> http://manning.com/ghosh
>
> Twttr: @debasishg
> Blog: http://debasishg.blogspot.com
> Code: http://github.com/debasishg
>
--
-- Guozhang
Re: facing serialization issues with Kafka Streams application ..
Posted by Debasish Ghosh <gh...@gmail.com>.
On Tue, Jun 20, 2017 at 7:03 PM, Damian Guy <da...@gmail.com> wrote:
> m.groupBy(mapper, Serdes.String(), Serdes.String()).count("
> HostAggregateCount")
>
Thanks! It works ..
--
Debasish Ghosh
http://manning.com/ghosh2
http://manning.com/ghosh
Twttr: @debasishg
Blog: http://debasishg.blogspot.com
Code: http://github.com/debasishg
Re: facing serialization issues with Kafka Streams application ..
Posted by Damian Guy <da...@gmail.com>.
Hi,
You should provide the serdes in the `groupByKey()` operation. The `map`
will trigger a re-partition in the `groupByKey` as you have changed the
key.
In fact you could replace the `map` and `groupByKey` with:
m.groupBy(mapper, Serdes.String(),
Serdes.String()).count("HostAggregateCount")
Thanks,
Damian
On Tue, 20 Jun 2017 at 12:08 Debasish Ghosh <gh...@gmail.com>
wrote:
> Hi -
>
> I have the following Scala snippet in a Kafka streams application ..
>
>
> val builder = new KStreamBuilder()
> val logRecords: KStream[Array[Byte], LogRecord] =
> builder.stream(Serdes.ByteArray(), logRecordSerde, config.toTopic)
> val m: KStream[Array[Byte], String] = logRecords.mapValues(hostExtractor)
> val n: KStream[String, String] = m.map(
> new KeyValueMapper[Array[Byte], String, KeyValue[String, String]]() {
> override def apply(key: Array[Byte], value: String): KeyValue[String,
> String] = new KeyValue(value, value)
> })
>
> val counts: KTable[String, java.lang.Long] =
> n.groupByKey().count("HostAggregateCount")
>
>
> In the above snippet logRecordSerde is a custom implementation of a Serde
> based on a Scala case class.
>
> When I run this application, I get the following exception ..
>
> Exception in thread "StreamThread-1"
> > org.apache.kafka.streams.errors.StreamsException: Exception caught in
> > process. taskId=0_0, processor=KSTREAM-SOURCE-0000000000,
> > topic=processed-log, partition=0, offset=0
> > at
> >
> org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:216)
> > at
> >
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:627)
> > at
> >
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:361)
> > Caused by: org.apache.kafka.streams.errors.StreamsException: *A
> > serializer (key:
> org.apache.kafka.common.serialization.ByteArraySerializer
> > / value: org.apache.kafka.common.serialization.ByteArraySerializer) is
> not
> > compatible to the actual key or value type (key type: java.lang.String /
> > value type: java.lang.String).* Change the default Serdes in StreamConfig
> > or provide correct Serdes via method parameters.
> > at
> >
> org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:81)
> > at
> >
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:83)
>
>
> Possibly the exception comes when I try to do a map on the KStream, which
> is supposed to return KStream[String, String]. My question is how do I
> inject the proper Serde here ? map does not take any Serde as arguments.
> The initial topic from where I read has ByteArray and a custom Serde and it
> works fine. But then the map blows up ..
>
> Any help will be appreciated ..
>
> regards.
>
>
>
> --
> Debasish Ghosh
> http://manning.com/ghosh2
> http://manning.com/ghosh
>
> Twttr: @debasishg
> Blog: http://debasishg.blogspot.com
> Code: http://github.com/debasishg
>