You are viewing a plain text version of this content. The canonical link for it is here.

Posted to user@spark.apache.org by Ashish Soni <as...@gmail.com> on 2015/06/26 14:39:48 UTC

Kafka Direct Stream - Custom Serialization and Deserilization

Hi ,

If i have a below data format , how can i use kafka direct stream to
de-serialize as i am not able to understand all the parameter i need to
pass , Can some one explain what will be the arguments as i am not clear
about this

JavaPairInputDStream
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82org.apache.spark.streaming.api.java.JavaPairInputDStream>
<K
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82K>,
V
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82V>>
org
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg>
.apache
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache>
.spark
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark>
.streaming
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming>
.kafka
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka>
.KafkaUtils
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils>
.createDirectStream(JavaStreamingContext
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82org.apache.spark.streaming.api.java.JavaStreamingContext>
arg0, Class
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82java.lang.Class>
<K
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82K>>
arg1, Class
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82java.lang.Class>
<V
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82V>>
arg2, Class
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82java.lang.Class>
<KD
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82KD>>
arg3, Class
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82java.lang.Class>
<VD
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82VD>>
arg4, Map
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82java.util.Map>
<String
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82java.lang.String>,
String
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82java.lang.String>>
arg5, Set
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82java.util.Set>
<String
<eclipse-javadoc:%E2%98%82=cde/C:%5C/Users%5C/eassoni%5C/.m2%5C/repository%5C/org%5C/apache%5C/spark%5C/spark-streaming-kafka_2.10%5C/1.4.0%5C/spark-streaming-kafka_2.10-1.4.0.jar%3Corg.apache.spark.streaming.kafka(KafkaUtils.class%E2%98%83KafkaUtils~createDirectStream~Lorg.apache.spark.streaming.api.java.JavaStreamingContext;~Ljava.lang.Class%5C%3CTK;%3E;~Ljava.lang.Class%5C%3CTV;%3E;~Ljava.lang.Class%5C%3CTKD;%3E;~Ljava.lang.Class%5C%3CTVD;%3E;~Ljava.util.Map%5C%3CLjava.lang.String;Ljava.lang.String;%3E;~Ljava.util.Set%5C%3CLjava.lang.String;%3E;%E2%98%82java.lang.String>>
arg6)

ID
Name
Unit
Rate
Duration

Re: Kafka Direct Stream - Custom Serialization and Deserilization

Posted by Benjamin Fradet <be...@gmail.com>.

There is one for the key of your Kafka message and one for its value.
On 26 Jun 2015 4:21 pm, "Ashish Soni" <as...@gmail.com> wrote:

> my question is why there are similar two parameter String.Class and
> StringDecoder.class what is the difference each of them ?
>
> Ashish
>
> On Fri, Jun 26, 2015 at 8:53 AM, Akhil Das <ak...@sigmoidanalytics.com>
> wrote:
>
>> JavaPairInputDStream<String, String> messages =
>> KafkaUtils.createDirectStream(
>>         jssc,
>>         String.class,
>>         String.class,
>>         StringDecoder.class,
>>         StringDecoder.class,
>>         kafkaParams,
>>         topicsSet
>>     );
>>
>> Here:
>>
>> jssc => JavaStreamingContext
>> String.class => Key , Value classes
>> StringDecoder => Key, Value decoder classes
>> KafkaParams => Map in which you specify all the kafka details (like
>> brokers, offset etc)
>> topicSet => Set of topics from which you want to consume data.
>>
>> Here's a sample program
>> <https://github.com/apache/spark/blob/master/examples/src/main/java/org/apache/spark/examples/streaming/JavaDirectKafkaWordCount.java>
>> for you to start.
>>
>>
>>
>> Thanks
>> Best Regards
>>
>> On Fri, Jun 26, 2015 at 6:09 PM, Ashish Soni <as...@gmail.com>
>> wrote:
>>
>>> Hi ,
>>>
>>> If i have a below data format , how can i use kafka direct stream to
>>> de-serialize as i am not able to understand all the parameter i need to
>>> pass , Can some one explain what will be the arguments as i am not clear
>>> about this
>>>
>>> JavaPairInputDStream<K, V> org.apache.spark.streaming.kafka.KafkaUtils
>>> .createDirectStream(JavaStreamingContext arg0, Class<K> arg1, Class<V>
>>> arg2, Class<KD> arg3, Class<VD> arg4, Map<String, String> arg5, Set<
>>> String> arg6)
>>>
>>> ID
>>> Name
>>> Unit
>>> Rate
>>> Duration
>>>
>>>
>>
>

Re: Kafka Direct Stream - Custom Serialization and Deserilization

Posted by Ashish Soni <as...@gmail.com>.

my question is why there are similar two parameter String.Class and
StringDecoder.class what is the difference each of them ?

Ashish

On Fri, Jun 26, 2015 at 8:53 AM, Akhil Das <ak...@sigmoidanalytics.com>
wrote:

> JavaPairInputDStream<String, String> messages =
> KafkaUtils.createDirectStream(
>         jssc,
>         String.class,
>         String.class,
>         StringDecoder.class,
>         StringDecoder.class,
>         kafkaParams,
>         topicsSet
>     );
>
> Here:
>
> jssc => JavaStreamingContext
> String.class => Key , Value classes
> StringDecoder => Key, Value decoder classes
> KafkaParams => Map in which you specify all the kafka details (like
> brokers, offset etc)
> topicSet => Set of topics from which you want to consume data.
>
> Here's a sample program
> <https://github.com/apache/spark/blob/master/examples/src/main/java/org/apache/spark/examples/streaming/JavaDirectKafkaWordCount.java>
> for you to start.
>
>
>
> Thanks
> Best Regards
>
> On Fri, Jun 26, 2015 at 6:09 PM, Ashish Soni <as...@gmail.com>
> wrote:
>
>> Hi ,
>>
>> If i have a below data format , how can i use kafka direct stream to
>> de-serialize as i am not able to understand all the parameter i need to
>> pass , Can some one explain what will be the arguments as i am not clear
>> about this
>>
>> JavaPairInputDStream<K, V> org.apache.spark.streaming.kafka.KafkaUtils
>> .createDirectStream(JavaStreamingContext arg0, Class<K> arg1, Class<V>
>> arg2, Class<KD> arg3, Class<VD> arg4, Map<String, String> arg5, Set<
>> String> arg6)
>>
>> ID
>> Name
>> Unit
>> Rate
>> Duration
>>
>>
>

Re: Kafka Direct Stream - Custom Serialization and Deserilization

Posted by Akhil Das <ak...@sigmoidanalytics.com>.

JavaPairInputDStream<String, String> messages =
KafkaUtils.createDirectStream(
        jssc,
        String.class,
        String.class,
        StringDecoder.class,
        StringDecoder.class,
        kafkaParams,
        topicsSet
    );

Here:

jssc => JavaStreamingContext
String.class => Key , Value classes
StringDecoder => Key, Value decoder classes
KafkaParams => Map in which you specify all the kafka details (like
brokers, offset etc)
topicSet => Set of topics from which you want to consume data.

Here's a sample program
<https://github.com/apache/spark/blob/master/examples/src/main/java/org/apache/spark/examples/streaming/JavaDirectKafkaWordCount.java>
for you to start.

Thanks
Best Regards

On Fri, Jun 26, 2015 at 6:09 PM, Ashish Soni <as...@gmail.com> wrote:

> Hi ,
>
> If i have a below data format , how can i use kafka direct stream to
> de-serialize as i am not able to understand all the parameter i need to
> pass , Can some one explain what will be the arguments as i am not clear
> about this
>
> JavaPairInputDStream<K, V> org.apache.spark.streaming.kafka.KafkaUtils
> .createDirectStream(JavaStreamingContext arg0, Class<K> arg1, Class<V>
> arg2, Class<KD> arg3, Class<VD> arg4, Map<String, String> arg5, Set<String>
> arg6)
>
> ID
> Name
> Unit
> Rate
> Duration
>
>