Getting Empty set while reading data from kafka-Spark-Streaming

Question

Hi i am new to Spark Streaming. i am trying to read the xml file and send it to kafka topic. Here is my Kafka Code Which sends data to Kafka-console-consumer.

Code:

package org.apache.kafka.Kafka_Producer;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.Properties;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutionException;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

@SuppressWarnings("unused")
public class KafkaProducer { 
   private static String sCurrentLine;
   public static void main(String args[]) throws InterruptedException, ExecutionException{ 
       try (BufferedReader br = new BufferedReader(new FileReader("/Users/sreeharsha/Downloads/123.txt")))
       {
           while ((sCurrentLine = br.readLine()) != null) {
               System.out.println(sCurrentLine);
               kafka(sCurrentLine);
           }
       } catch (FileNotFoundException e) {
           // TODO Auto-generated catch block
           e.printStackTrace();
       } catch (IOException e) {
           // TODO Auto-generated catch block
           e.printStackTrace();}
   }
   public static void kafka(String sCurrentLine)  {
       Properties props = new Properties();
       props.put("metadata.broker.list", "localhost:9092");
       props.put("serializer.class", "kafka.serializer.StringEncoder");
       props.put("partitioner.class","kafka.producer.DefaultPartitioner");
       props.put("request.required.acks", "1");
       ProducerConfig config = new ProducerConfig(props);
       Producer<String, String> producer = new Producer<String, String>(config);
       producer.send(new KeyedMessage<String, String>("sample",sCurrentLine));
       producer.close();
   }
}

i can recieve the data in Kafka-Console-Consumer. In the below screenshot you can see the data which i had sent to the topic.

Now i need to stream the data which i send to kafka-console-consumer using Spark-Streaming. Here is the code.

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

public class SparkStringConsumer {

   public static void main(String[] args) {

       SparkConf conf = new SparkConf()
               .setAppName("kafka-sandbox")
               .setMaster("local[*]");
       JavaSparkContext sc = new JavaSparkContext(conf);
       JavaStreamingContext ssc = new JavaStreamingContext(sc, new Duration(2000));

       Map<String, String> kafkaParams = new HashMap<>();
       kafkaParams.put("metadata.broker.list", "localhost:9092");
       Set<String> topics = Collections.singleton("sample");

       JavaPairInputDStream<String, String> directKafkaStream = KafkaUtils.createDirectStream(ssc,
       String.class, String.class, StringDecoder.class, StringDecoder.class, kafkaParams, topics);
       directKafkaStream.foreachRDD(rdd -> {
       System.out.println("--- New RDD with " + rdd.partitions().size()
           + " partitions and " + rdd.count() + " records");
       rdd.foreach(record -> System.out.println(record._2));
       });
       ssc.start();
       ssc.awaitTermination();
   }
}

Getting emptyset while submitting my job like this:

./spark-submit --class org.apache.spark_streaming.Spark_Kafka_Streaming.SparkStringConsumer --master local[4] Spark_Kafka_Streaming-0.0.1-SNAPSHOT.jar

Below you can see the screenshot of how the data is receiving:

Using the below versions:

Spark - 2.0.0

Zookeeper -3.4.6

Kafka - 0.8.2.1

Any suggestions please,

Where is the code for SparkReceiver class? You have posted SparkStringConsumer class in which you are using topic as "mytopic" and in KafkaProducer class you are sending messages on topic "sample". Can you please check? — abaghel
Getting messages to kafka its not a problem while reading getting empty set in spark — Sree Eedupuganti
Try to use this in your producer class instead of reading from file. This is for test. Random random = new Random(); while(true){ kafka("Test- "+random.nextInt(100)); Thread.sleep(500); } } Please also check from where it is resolving StringDecoder.class in SparkStringConsumer class. It should be import kafka.serializer.StringDecoder; — abaghel

Sree Eedupuganti Sree Eedupuganti · Accepted Answer · 2016-09-20T07:00:25

Finally after surfing over internet i found these solution.

Don't use "Spark-Submit" and "SetMaster" at the same time.

If you run the code from your IDE, use SetMaster in your code
If you run the jar through "Spark-Submit" do not put setMaster in your code

And one more thing first run/submit your spark jar and then send the data to Kafka-Console-Consumer

Working Fine.

Getting Empty set while reading data from kafka-Spark-Streaming

1 Answers