Spark Ausführung für Twitter Streaming

Question

Hallo, ich bin neu zu funken und scala. Ich versuche, einige Tweets durch Funken mit dem folgenden Code-Streaming zu streamen:

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) 

Mein Fehler:

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) 

kann jemand überprüfen Sie den Code und mir sagen, wo ich falsch mache?

Answer 1

Wenn Ihre Konfigurationsdatei enthält:

SPARK_MASTER=local[2] 

Ändern Sie es an:

SPARK_MASTER="local[2]"