如何使用scala运行Spark流的twitter流行标签？

我是Spark新手，请指导。如何使用scala运行Spark流的twitter流行标签？

有很多与Spark使用Scala相关的示例。

您可以从https://github.com/apache/incubator-spark/tree/master/examples/src/main/scala/org/apache/spark/streaming/examples中查看。

我想运行TwitterPopularTags.scala。

我无法为此示例设置twitter登录详细信息。

http://spark.incubator.apache.org/docs/latest/streaming-programming-guide.html#linking-with-spark-streaming

我成功地运行网络计数的例子。

但是，当我执行
./run-example org.apache.spark.streaming.examples.TwitterPopularTags local[2] 那么它会告诉我认证失败的问题...

我TwitterPopularTags.scala初始化字符串情形类似

System.setProperty("twitter4j.oauth.consumerKey", "####"); 
System.setProperty("twitter4j.oauth.consumerSecret", "##"); 
System.setProperty("twitter4j.oauth.accessToken", "##"); 
System.setProperty("twitter4j.oauth.accessTokenSecret", "##");

请指导之前设置Twitter的登录信息。

来源

2013-10-30 Parag Kuhikar

http://spark.incubator.apache.org/docs/latest/streaming-programming-guide.html#linking-with-spark-streaming 我成功运行网络计数示例。但是，当我执行./run-example org.apache.spark.streaming.examples.TwitterPopularTags本地[2]然后它会显示我身份验证失败问题... 我设置了t –

您可以请编辑您的问题，包括您收到的特定错误消息或异常？另外，AMP Camp 3大数据迷你课程有一个流式Twitter示例，可能会有所帮助（并且包含有关生成相应API令牌的说明）：http://ampcamp.berkeley.edu/big-data-mini-course/realtime - 处理与 - 火花streaming.html –

在运行Twitter示例之前，将文件“twitter4j.properties”放入Spark根目录（例如spark-0.8.0-incubating）。

twitter4j.properties：

oauth.consumerKey=*** 
oauth.consumerSecret=*** 
oauth.accessToken=*** 
oauth.accessTokenSecret=***

为我工作在Mac上使用Scala的例子。

来源

2013-11-01 17:50:51 Dronkel

我无法打开github链接https://github.com/apache/incubator-spark/tree/master/examples/src/main/scala/org/apache/spark/streaming/examples。

但是，您可以使用下面的代码为我工作。

import org.apache.spark.streaming.{ Seconds, StreamingContext } 
import org.apache.spark.SparkContext._ 
import org.apache.spark.streaming.twitter._ 
import org.apache.spark.SparkConf 
import org.apache.spark.streaming._ 
import org.apache.spark.{ SparkContext, SparkConf } 
import org.apache.spark.storage.StorageLevel 
import org.apache.spark.streaming.flume._ 

/** 
* A Spark Streaming application that receives tweets on certain 
* keywords from twitter datasource and find the popular hashtags 
* 
* Arguments: <comsumerKey> <consumerSecret> <accessToken> <accessTokenSecret> <keyword_1> ... <keyword_n> 
* <comsumerKey>  - Twitter consumer key 
* <consumerSecret>  - Twitter consumer secret 
* <accessToken>  - Twitter access token 
* <accessTokenSecret> - Twitter access token secret 
* <keyword_1>   - The keyword to filter tweets 
* <keyword_n>   - Any number of keywords to filter tweets 
* 
* More discussion at stdatalabs.blogspot.com 
* 
* @author Sachin Thirumala 
*/ 

object SparkPopularHashTags { 
    val conf = new SparkConf().setMaster("local[4]").setAppName("Spark Streaming - PopularHashTags") 
    val sc = new SparkContext(conf) 

    def main(args: Array[String]) { 

    sc.setLogLevel("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 generat 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) 

    // Set the Spark StreamingContext to create a DStream for every 5 seconds 
    val ssc = new StreamingContext(sc, Seconds(5)) 
    // Pass the filter keywords as arguements 

    // val stream = FlumeUtils.createStream(ssc, args(0), args(1).toInt) 
    val stream = TwitterUtils.createStream(ssc, None, filters) 

    // Split the stream on space and extract hashtags 
    val hashTags = stream.flatMap(status => status.getText.split(" ").filter(_.startsWith("#"))) 

    // Get the top hashtags over the previous 60 sec window 
    val topCounts60 = hashTags.map((_, 1)).reduceByKeyAndWindow(_ + _, Seconds(60)) 
     .map { case (topic, count) => (count, topic) } 
     .transform(_.sortByKey(false)) 

    // Get the top hashtags over the previous 10 sec window 
    val topCounts10 = hashTags.map((_, 1)).reduceByKeyAndWindow(_ + _, Seconds(10)) 
     .map { case (topic, count) => (count, topic) } 
     .transform(_.sortByKey(false)) 

    // print tweets in the currect DStream 
    stream.print() 

    // 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() 
    } 
}

说明：
setMaster("local[4]") - 确保为1个线程用于收集流入流束和另一个线程用于处理它掌握与至少2个线程设置为本地模式。

我们指望的流行主题标签与下面的代码：

val topCounts60 = hashTags.map((_, 1)).reduceByKeyAndWindow(_ + _, Seconds(60)) 
     .map { case (topic, count) => (count, topic) } 
     .transform(_.sortByKey(false))

上面的代码做了主题标签的字数超过作为reduceByKeyAndWindow指定的上一个60/10秒，并以降序排序。

reduceByKeyAndWindow用于我们必须对先前流间隔中累积的数据应用转换的情况。

通过将四个叽叽喳喳OAuth凭证作为参数执行代码：