You are not having the twitter4j jars in the classpath. While running it in
the cluster mode you need to ship those dependency jars.
You can do like:
sparkConf.setJars("/home/akhld/jars/twitter4j-core-3.0.3.jar,
/home/akhld/jars/twitter4j-stream-3.0.3.jar")
You can make sure they are shipped by checking the Application WebUI (4040)
environment tab.
Thanks
Best Regards
On Tue, Nov 11, 2014 at 5:48 PM, Jishnu Menath Prathap (WT01 - BAS) <
jishnu.prat...@wipro.com> wrote:
> *Hi I am getting the following error while executing a scala_twitter
> program for spark*
> 14/11/11 16:39:23 ERROR receiver.ReceiverSupervisorImpl: Stopped executor
> with error: java.lang.NoSuchMethodError:
> twitter4j.TwitterStream.addListener(Ltwitter4j/StatusListener;)V
> 14/11/11 16:39:23 ERROR executor.Executor: Exception in task 0.0 in stage
> 0.0 (TID 0)
> java.lang.NoSuchMethodError:
> twitter4j.TwitterStream.addListener(Ltwitter4j/StatusListener;)V
> at
> org.apache.spark.streaming.twitter.TwitterReceiver.onStart(TwitterInputDStream.scala:72)
>
> at
> org.apache.spark.streaming.receiver.ReceiverSupervisor.startReceiver(ReceiverSupervisor.scala:121)
>
> at
> org.apache.spark.streaming.receiver.ReceiverSupervisor.start(ReceiverSupervisor.scala:106)
>
> at
> org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverLauncher$$anonfun$9.apply(ReceiverTracker.scala:264)
>
> at
> org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverLauncher$$anonfun$9.apply(ReceiverTracker.scala:257)
>
> at
> org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1121)
>
> at
> org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1121)
>
> at
> org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:62)
> at org.apache.spark.scheduler.Task.run(Task.scala:54)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:177)
> at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> Source)
> at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown
> Source)
> at java.lang.Thread.run(Unknown Source)
> 14/11/11 16:39:23 ERROR executor.ExecutorUncaughtExceptionHandler:
> Uncaught exception in thread Thread[Executor task launch worker-0,5,main]
> java.lang.NoSuchMethodError:
> twitter4j.TwitterStream.addListener(Ltwitter4j/StatusListener;)V
> at
> org.apache.spark.streaming.twitter.TwitterReceiver.onStart(TwitterInputDStream.scala:72)
>
> at
> org.apache.spark.streaming.receiver.ReceiverSupervisor.startReceiver(ReceiverSupervisor.scala:121)
>
> at
> org.apache.spark.streaming.receiver.ReceiverSupervisor.start(ReceiverSupervisor.scala:106)
>
> at
> org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverLauncher$$anonfun$9.apply(ReceiverTracker.scala:264)
>
> at
> org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverLauncher$$anonfun$9.apply(ReceiverTracker.scala:257)
>
> at
> org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1121)
>
> at
> org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1121)
>
> at
> org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:62)
> at org.apache.spark.scheduler.Task.run(Task.scala:54)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:177)
> at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> Source)
> at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown
> Source)
> at java.lang.Thread.run(Unknown Source)
> 1
> *I think it might be a dependency issue so sharing pom.xml too*
>
> <dependencies>
> <dependency>
> <groupId>org.twitter4j</groupId>
> <artifactId>twitter4j-core</artifactId>
> <version>3.0.3</version>
> </dependency>
> <dependency>
> <groupId>org.apache.httpcomponents</groupId>
> <artifactId>httpclient</artifactId>
> <version>4.0-beta1</version>
> </dependency>
> <dependency>
> <groupId>org.apache.httpcomponents</groupId>
> <artifactId>httpclient</artifactId>
> <version>4.3.5</version>
> </dependency>
> <dependency>
> <groupId>oauth.signpost</groupId>
> <artifactId>signpost-commonshttp4</artifactId>
> <version>1.2</version>
> </dependency>
> <dependency>
> <groupId>org.scalatest</groupId>
> <artifactId>scalatest_2.10</artifactId>
> <version>3.0.0-SNAP2</version>
> </dependency>
> <dependency>
> <groupId>commons-io</groupId>
> <artifactId>commons-io</artifactId>
> <version>2.4</version>
> </dependency>
> <dependency>
> <groupId>junit</groupId>
> <artifactId>junit</artifactId>
> <version>4.4</version>
> </dependency>
> <dependency>
> <groupId>org.twitter4j</groupId>
> <artifactId>twitter4j-stream</artifactId>
> <version>3.0.3</version>
> </dependency>
> <dependency>
> <groupId>org.apache.spark</groupId>
> <artifactId>spark-streaming_2.10</artifactId>
> <version>1.1.0</version>
> </dependency>
> <dependency>
> <groupId>org.apache.spark</groupId>
> <artifactId>spark-core_2.10</artifactId>
> <version>1.1.0</version>
> </dependency>
> <dependency>
> <groupId>org.apache.spark</groupId>
>
> <artifactId>spark-streaming-twitter_2.10</artifactId>
> <version>1.1.0</version>
> </dependency>
>
> </dependencies>
>
> * Code of the twitter streaming spark *
>
> object sparktwitteruser {
> def main(args: Array[String]) {
>
>
> val filters = Array("hi")
> // Set the system properties so
> that Twitter4j library used by twitter stream
> // can use them to generat OAuth
> credentials
>
>
> System.setProperty("twitter4j.oauth.consumerKey", "******")
>
> System.setProperty("twitter4j.oauth.consumerSecret","*****")
>
> System.setProperty("twitter4j.oauth.accessToken","*******")
>
> System.setProperty("twitter4j.oauth.accessTokenSecret", "*******")
>
> val sparkConf = new
> SparkConf().setAppName("TwitterPopularTags").setMaster("local").set("spark.eventLog.enabled","true")
>
> val ssc = new
> StreamingContext(sparkConf, Seconds(2))
> val stream =
> TwitterUtils.createStream(ssc, None, filters)
> val hashTags =
> stream.flatMap(status => status.getText.split("
> ").filter(_.startsWith("#")))
> val topCounts60 = hashTags.map((_,
> 1)).reduceByKeyAndWindow(_ + _, Seconds(60))
> .map{case (topic, count) =>
> (count, topic)}
> .transform(_.sortByKey(false))
>
> val topCounts10 = hashTags.map((_,
> 1)).reduceByKeyAndWindow(_ + _, Seconds(10))
> .map{case (topic, count) =>
> (count, topic)}
> .transform(_.sortByKey(false))
> // Print popular hashtags
>
> topCounts60.foreachRDD(rdd => {
> val topList = rdd.take(10)
>
> println("\nPopular topics in last 60 seconds (%s
> total):".format(rdd.count()))
>
> topList.foreach{case (count, tag) => println("%s (%s tweets)".format(tag,
> count))}
> })
>
> topCounts10.foreachRDD(rdd => {
> val topList = rdd.take(10)
>
> println("\nPopular topics in last 10 seconds (%s
> total):".format(rdd.count()))
>
> topList.foreach{case (count, tag) => println("%s (%s tweets)".format(tag,
> count))}
> })
>
> ssc.start()
> ssc.awaitTermination()
> }
> }
>
> The information contained in this electronic message and any attachments
> to this message are intended for the exclusive use of the addressee(s) and
> may contain proprietary, confidential or privileged information. If you are
> not the intended recipient, you should not disseminate, distribute or copy
> this e-mail. Please notify the sender immediately and destroy all copies of
> this message and any attachments.
>
> WARNING: Computer viruses can be transmitted via email. The recipient
> should check this email and any attachments for the presence of viruses.
> The company accepts no liability for any damage caused by any virus
> transmitted by this email.
>
> www.wipro.com
>
