Hi Team
I have been struggling since a week now trying to read popular tags from Twitter in my Spark Application.
My Environment is as below that is a setup from our Organization's Client
Cdh 5.14
Spark 1.6
Scala 2.10
Below is the error i am getting ,pasting the complete log for reference .
and i have been trying the below to solve it but with no avail.
20/01/25 18:13:00 WARN scheduler.TaskSetManager: Lost task 0.0 in stage 3.0 (TID 72, tplhc01d191.iuser.iroot.adidom.com, executor 1): java.lang.NoSuchMethodError: twitter4j.TwitterStream.addListener(Ltwitter4j/StreamListener;)V at org.apache.spark.streaming.twitter.TwitterReceiver.onStart(TwitterInputDStream.scala:72) at org.apache.spark.streaming.receiver.ReceiverSupervisor.startReceiver(ReceiverSupervisor.scala:148) at org.apache.spark.streaming.receiver.ReceiverSupervisor.start(ReceiverSupervisor.scala:130) at org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverTrackerEndpoint$$anonfun$9.apply(ReceiverTracker.scala:575) at org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverTrackerEndpoint$$anonfun$9.apply(ReceiverTracker.scala:565) at org.apache.spark.SparkContext$$anonfun$38.apply(SparkContext.scala:2022) at org.apache.spark.SparkContext$$anonfun$38.apply(SparkContext.scala:2022) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) at org.apache.spark.scheduler.Task.run(Task.scala:89) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:242) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:744)
20/01/25 18:13:01 ERROR cluster.YarnScheduler: Lost executor 2 on tplhc01d191.iuser.iroot.adidom.com: Container marked as failed: container_e168_1579122798111_703147_01_000003 on host: tplhc01d191.iuser.iroot.adidom.com. Exit status: 50. Diagnostics: Exception from container-launch. Container id: container_e168_1579122798111_703147_01_000003 Exit code: 50 Stack trace: ExitCodeException exitCode=50: at org.apache.hadoop.util.Shell.runCommand(Shell.java:604) at org.apache.hadoop.util.Shell.run(Shell.java:507) at org.apache.hadoop.util.Shell$ShellCommandExecutor.execute(Shell.java:789) at org.apache.hadoop.yarn.server.nodemanager.LinuxContainerExecutor.launchContainer(LinuxContainerExecutor.java:399) at org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch.call(ContainerLaunch.java:302) at org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch.call(ContainerLaunch.java:82) at java.util.concurrent.FutureTask.run(FutureTask.java:262) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:744)
Shell output: main : command provided 1 main : run as user is ZB609239 main : requested yarn user is ZB609239 Writing to tmp file /data/14/yarn/nm/nmPrivate/application_1579122798111_703147/container_e168_1579122798111_703147_01_000003/container_e168_1579122798111_703147_01_000003.pid.tmp Writing to cgroup task files...
Container exited with a non-zero exit code 50
My Command at execution is as below
spark-submit --queue NONP.HAASXXXX_XXX --conf spark.ui.port=0 --jars "/home/XXXXX/spark-streaming-twitter_2.10-1.6.1.jar","/home/XXXXX/twitter4j-core-4.0.4.jar","/home/XXXXX/twitter4j-stream-4.0.4.jar" --class com.sparkStreaming.TwitterTags TwitterTagsBT.jar XXXX XXXXXRR RRRREEXX WWEERRDD twittertagfilter --conf spark.yarn.user.classpath.first=true --conf spark.shuffle.memoryFraction=0.5 --conf spark.yarn.executor.memoryOverhead=8192 --conf spark.akka.timeout=300 --conf spark.storage.blockManagerSlaveTimeoutMs=300000 --conf spark.yarn.driver.memoryOverhead=2048 --executor-memory 25g --driver-memory 4g --num-executors 45 --executor-cores 5 --conf spark.executor.extrajavaoptions="-XX:-UseGCOverheadLimit" --conf spark.executor.userClassPathFirst=true --conf spark.driver.userClassPathFirst=truels
My Code goes as below.Just to add i have tried with twitter4j 3.0.6,3.0.3 and 4.0.7 versions as well,for all of them i get the same above error and it doesn't connect.Please advise what would be the problem here.
package com.sparkStreaming
import org.apache.spark.streaming._ import org.apache.spark.sql._ import org.apache.spark.sql.hive.orc._ import org.apache.spark.sql.expressions.Window import org.apache.spark.sql.functions._ import org.apache.spark.rdd.RDD import org.apache.spark.sql.Row import org.apache.spark._ import org.apache.spark.SparkContext._ import java.nio.charset.CodingErrorAction import scala.io.Codec import org.apache.spark.storage.StorageLevel import org.apache.spark._ import org.apache.spark.util.Utils import org.apache.spark.sql.Dataset import org.apache.spark.sql.expressions import org.apache.spark.sql.functions.{ concat, lit } import org.apache.spark.sql.types._ import org.apache.spark.sql.functions.udf import java.util.Calendar import org.apache.hadoop.fs.FileSystem import org.apache.hadoop.fs.Path import org.apache.hadoop.security.UserGroupInformation.HadoopConfiguration import org.apache.log4j._ import org.apache.spark.streaming.twitter.TwitterUtils import org.apache.spark.streaming.twitter._ import twitter4j.Status; import twitter4j.auth.Authorization import twitter4j.auth.OAuthAuthorization import twitter4j.conf.Configuration import twitter4j.conf.ConfigurationBuilder
object TwitterTags {
def main(args: Array[String]) { if (args.length < 3) { System.err.println("Usage: TwitterSentiments <consumer key> <consumer secret> " + "<access token> <access token secret> [<filters>]") System.exit(1) } // Set logging level if log4j not configured (override by adding log4j.properties to classpath) if (!Logger.getRootLogger.getAllAppenders.hasMoreElements) { Logger.getRootLogger.setLevel(Level.WARN) } val Array(consumerKey, consumerSecret, accessToken, accessTokenSecret) = args.take(4) //val filters = args.takeRight(args.length - 4) // Set the system properties so that Twitter4j library used by Twitter stream // can use them to generate OAuth credentials System.setProperty("twitter4j.oauth.consumerKey", consumerKey) System.setProperty("twitter4j.oauth.consumerSecret", consumerSecret) System.setProperty("twitter4j.oauth.accessToken", accessToken) System.setProperty("twitter4j.oauth.accessTokenSecret", accessTokenSecret) val sparkConf = new SparkConf().setAppName("TwitterHashTagJoinSentiments"). setJars(Array("/home/zb609239/twitter4j-core-4.0.4.jar,/home/zb609239/twitter4j-stream-4.0.4.jar,spark-streaming-twitter_2.10-1.6.1.jar")) val ssc = new StreamingContext(sparkConf, Seconds(2)) ssc.sparkContext.setLogLevel("WARN") val stream = TwitterUtils.createStream(ssc, None, filters) val stream = TwitterUtils.createStream(ssc,None) val hashtags = stream.flatMap(status => status.getText.split(" ").filter(_.startsWith("@")))
val topCounts60 = hashtags.map(hashTag => (hashTag, 1)).reduceByKeyAndWindow(_ + _, Seconds(86400)) .map { case (topic, count) => (count, topic) } .transform(_.sortByKey(false))
topCounts60.foreachRDD(rdd => { val topList = rdd.take(10) println("\nLatest topics in last 24 hours (%s total):".format(rdd.count())) topList.foreach { case (count, tag) => println("%s (%s tweets)".format(tag, count)) } })
ssc.start() ssc.awaitTermination()
}
}
... View more
