error while consuming data from kafka topic using JavaStreamingContext

Previous Topic Next Topic
 
classic Classic list List threaded Threaded
1 message Options
Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

error while consuming data from kafka topic using JavaStreamingContext

bhargavatanmay
This post has NOT been accepted by the mailing list yet.
i have been trying to consume data from a kafka topic and place it in hdfs but i have been getting this error.can someone please help.

Exception in thread "main" java.lang.NoSuchMethodError: org.apache.hadoop.security.UserGroupInformation.setConfiguration(Lorg/apache/hadoop/conf/Configuration;)V
        at org.apache.spark.deploy.SparkHadoopUtil.<init>(SparkHadoopUtil.scala:36)
        at org.apache.spark.deploy.SparkHadoopUtil$.<init>(SparkHadoopUtil.scala:109)
        at org.apache.spark.deploy.SparkHadoopUtil$.<clinit>(SparkHadoopUtil.scala)
        at org.apache.spark.SparkContext.<init>(SparkContext.scala:228)
        at org.apache.spark.streaming.StreamingContext$.createNewSparkContext(StreamingContext.scala:549)
        at org.apache.spark.streaming.StreamingContext.<init>(StreamingContext.scala:75)
        at org.apache.spark.streaming.api.java.JavaStreamingContext.<init>(JavaStreamingContext.scala:130)
        at com.zdatainc.rts.spark.SentimentAnalysis.main(SentimentAnalysis.java:38)


the code is as follows:

 public static void main(String[] args)
    {
        BasicConfigurator.configure();
        SparkConf conf = new SparkConf().setAppName("Twitter Sentiment Analysis");

        if (args.length > 0)
            conf.setMaster(args[0]);
        else
            conf.setMaster("local[2]");
       
        JavaStreamingContext ssc = new JavaStreamingContext(conf,new Duration(2000));

        Map<String, Integer> topicMap = new HashMap<String, Integer>();
        topicMap.put(KAFKA_TOPIC, KAFKA_PARALLELIZATION);

        JavaPairReceiverInputDStream<String, String> messages =
            KafkaUtils.createStream(
                ssc,
                Properties.getString("rts.spark.zkhosts"),
                "twitter.sentimentanalysis.kafka",
                topicMap);
       


these are my imports :

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.*;
import org.apache.spark.streaming.*;
import org.apache.spark.streaming.api.java.*;
import org.apache.spark.streaming.kafka.*;
Loading...