[SPARK-9057][STREAMING] Twitter example joining to static RDD of word sentiment values

Example of joining a static RDD of word sentiments to a streaming RDD of Tweets in order to demo the usage of the transform() method.

Author: Jeff L <sha0lin@alumni.carnegiemellon.edu>

Closes #8431 from Agent007/SPARK-9057.

1 files changed, 96 insertions, 0 deletions

diff --git a/examples/src/main/scala/org/apache/spark/examples/streaming/TwitterHashTagJoinSentiments.scala b/examples/src/main/scala/org/apache/spark/examples/streaming/TwitterHashTagJoinSentiments.scala
new file mode 100644
index 0000000000..0328fa81ea
--- /dev/null
+++ b/examples/src/main/scala/org/apache/spark/examples/streaming/TwitterHashTagJoinSentiments.scala
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// scalastyle:off println
+package org.apache.spark.examples.streaming
+
+import org.apache.spark.SparkConf
+import org.apache.spark.streaming.twitter.TwitterUtils
+import org.apache.spark.streaming.{Seconds, StreamingContext}
+
+/**
+ * Displays the most positive hash tags by joining the streaming Twitter data with a static RDD of
+ * the AFINN word list (http://neuro.imm.dtu.dk/wiki/AFINN)
+ */
+object TwitterHashTagJoinSentiments {
+  def main(args: Array[String]) {
+    if (args.length < 4) {
+      System.err.println("Usage: TwitterHashTagJoinSentiments <consumer key> <consumer secret> " +
+        "<access token> <access token secret> [<filters>]")
+      System.exit(1)
+    }
+
+    StreamingExamples.setStreamingLogLevels()
+
+    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")
+    val ssc = new StreamingContext(sparkConf, Seconds(2))
+    val stream = TwitterUtils.createStream(ssc, None, filters)
+
+    val hashTags = stream.flatMap(status => status.getText.split(" ").filter(_.startsWith("#")))
+
+    // Read in the word-sentiment list and create a static RDD from it
+    val wordSentimentFilePath = "data/streaming/AFINN-111.txt"
+    val wordSentiments = ssc.sparkContext.textFile(wordSentimentFilePath).map { line =>
+      val Array(word, happinessValue) = line.split("\t")
+      (word, happinessValue)
+    } cache()
+
+    // Determine the hash tags with the highest sentiment values by joining the streaming RDD
+    // with the static RDD inside the transform() method and then multiplying
+    // the frequency of the hash tag by its sentiment value
+    val happiest60 = hashTags.map(hashTag => (hashTag.tail, 1))
+      .reduceByKeyAndWindow(_ + _, Seconds(60))
+      .transform{topicCount => wordSentiments.join(topicCount)}
+      .map{case (topic, tuple) => (topic, tuple._1 * tuple._2)}
+      .map{case (topic, happinessValue) => (happinessValue, topic)}
+      .transform(_.sortByKey(false))
+
+    val happiest10 = hashTags.map(hashTag => (hashTag.tail, 1))
+      .reduceByKeyAndWindow(_ + _, Seconds(10))
+      .transform{topicCount => wordSentiments.join(topicCount)}
+      .map{case (topic, tuple) => (topic, tuple._1 * tuple._2)}
+      .map{case (topic, happinessValue) => (happinessValue, topic)}
+      .transform(_.sortByKey(false))
+
+    // Print hash tags with the most positive sentiment values
+    happiest60.foreachRDD(rdd => {
+      val topList = rdd.take(10)
+      println("\nHappiest topics in last 60 seconds (%s total):".format(rdd.count()))
+      topList.foreach{case (happiness, tag) => println("%s (%s happiness)".format(tag, happiness))}
+    })
+
+    happiest10.foreachRDD(rdd => {
+      val topList = rdd.take(10)
+      println("\nHappiest topics in last 10 seconds (%s total):".format(rdd.count()))
+      topList.foreach{case (happiness, tag) => println("%s (%s happiness)".format(tag, happiness))}
+    })
+
+    ssc.start()
+    ssc.awaitTermination()
+  }
+}
+// scalastyle:on println