CODEWITHSUNDEEP
scalaapache-sparktypesafe-config
Anji
Anji

Reputation: 305

Spark Execution for twitter Streaming

Hi I'm new to spark and scala . I'm trying to stream some tweets through spark streaming with the following code:

object TwitterStreaming {

  def main(args: Array[String]): Unit = {
    if (args.length < 1) {
      System.err.println("WrongUsage:   PropertiesFile, [<filters>]")
      System.exit(-1)
    }

    StreamingExamples.setStreaningLogLevels()
    val myConfigFile = args(0)
    val batchInterval_s = 1
    val fileConfig = ConfigFactory.parseFile(new File(myConfigFile))
    val appConf = ConfigFactory.load(fileConfig)  
    // Set the system properties so that Twitter4j library used by twitter stream
    // can use them to generate OAuth credentials

    System.setProperty("twitter4j.oauth.consumerKey", appConf.getString("consumerKey"))
    System.setProperty("twitter4j.oauth.consumerSecret", appConf.getString("consumerSecret"))
    System.setProperty("twitter4j.oauth.accessToken", appConf.getString("accessToken"))
    System.setProperty("twitter4j.oauth.accessTokenSecret", appConf.getString("accessTokenSecret"))

    val sparkConf = new SparkConf().setAppName("TwitterStreaming").setMaster(appConf.getString("SPARK_MASTER"))//local[2]

    val ssc = new StreamingContext(sparkConf, Seconds(batchInterval_s)) // creating spark streaming context
    val stream = TwitterUtils.createStream(ssc, None)
    val tweet_data = stream.map(status => TweetData(status.getId, "@" + status.getUser.getScreenName, status.getText.trim()))
    tweet_data.foreachRDD(rdd => {
      println(s"A sample of tweets I gathered over ${batchInterval_s}s: ${rdd.take(10).mkString(" ")} (total tweets fetched: ${rdd.count()})")
    })
  }

}

case class TweetData(id: BigInt, author: String, tweetText: String)

My Error:

Exception in thread "main" com.typesafe.config.ConfigException$WrongType:/WorkSpace/InputFiles/application.conf: 5: Cannot concatenate object or list with a non-object-or-list, ConfigString("local") and SimpleConfigList([2]) are not compatible
at com.typesafe.config.impl.ConfigConcatenation.join(ConfigConcatenation.java:116)

can any one check the the code and tell me where I'm doing wrong?

Upvotes: 0

Views: 136

Answers (1)

Tzach Zohar
Tzach Zohar

Reputation: 37822

If your config file contains:

SPARK_MASTER=local[2]

Change it to:

SPARK_MASTER="local[2]"

Upvotes: 1

Related Questions