[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 <sha0...@alumni.carnegiemellon.edu> Closes #8431 from Agent007/SPARK-9057. Project: http://git-wip-us.apache.org/repos/asf/bahir/repo Commit: http://git-wip-us.apache.org/repos/asf/bahir/commit/c6155756 Tree: http://git-wip-us.apache.org/repos/asf/bahir/tree/c6155756 Diff: http://git-wip-us.apache.org/repos/asf/bahir/diff/c6155756 Branch: refs/heads/master Commit: c61557569466855728f41c8180bbde8a78e2fb20 Parents: c25b799 Author: Jeff L <sha0...@alumni.carnegiemellon.edu> Authored: Fri Dec 18 15:06:54 2015 +0000 Committer: Sean Owen <so...@cloudera.com> Committed: Fri Dec 18 15:06:54 2015 +0000 ---------------------------------------------------------------------- .../JavaTwitterHashTagJoinSentiments.java | 180 +++++++++++++++++++ .../twitter/TwitterHashTagJoinSentiments.scala | 96 ++++++++++ 2 files changed, 276 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/bahir/blob/c6155756/streaming-twitter/examples/src/main/java/org/apache/spark/examples/streaming/twitter/JavaTwitterHashTagJoinSentiments.java ---------------------------------------------------------------------- diff --git a/streaming-twitter/examples/src/main/java/org/apache/spark/examples/streaming/twitter/JavaTwitterHashTagJoinSentiments.java b/streaming-twitter/examples/src/main/java/org/apache/spark/examples/streaming/twitter/JavaTwitterHashTagJoinSentiments.java new file mode 100644 index 0000000..030ee30 --- /dev/null +++ b/streaming-twitter/examples/src/main/java/org/apache/spark/examples/streaming/twitter/JavaTwitterHashTagJoinSentiments.java @@ -0,0 +1,180 @@ +/* + * 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. + */ + +package org.apache.spark.examples.streaming; + +import org.apache.commons.io.IOUtils; +import org.apache.spark.SparkConf; +import org.apache.spark.api.java.JavaPairRDD; +import org.apache.spark.api.java.function.FlatMapFunction; +import org.apache.spark.api.java.function.Function; +import org.apache.spark.api.java.function.Function2; +import org.apache.spark.api.java.function.PairFunction; +import org.apache.spark.streaming.Duration; +import org.apache.spark.streaming.api.java.JavaDStream; +import org.apache.spark.streaming.api.java.JavaPairDStream; +import org.apache.spark.streaming.api.java.JavaReceiverInputDStream; +import org.apache.spark.streaming.api.java.JavaStreamingContext; +import org.apache.spark.streaming.twitter.TwitterUtils; +import scala.Tuple2; +import twitter4j.Status; + +import java.io.IOException; +import java.net.URI; +import java.util.Arrays; +import java.util.List; + +/** + * 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) + */ +public class JavaTwitterHashTagJoinSentiments { + + public static void main(String[] args) throws IOException { + if (args.length < 4) { + System.err.println("Usage: JavaTwitterHashTagJoinSentiments <consumer key> <consumer secret>" + + " <access token> <access token secret> [<filters>]"); + System.exit(1); + } + + StreamingExamples.setStreamingLogLevels(); + + String consumerKey = args[0]; + String consumerSecret = args[1]; + String accessToken = args[2]; + String accessTokenSecret = args[3]; + String[] filters = Arrays.copyOfRange(args, 4, args.length); + + // 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); + + SparkConf sparkConf = new SparkConf().setAppName("JavaTwitterHashTagJoinSentiments"); + JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(2000)); + JavaReceiverInputDStream<Status> stream = TwitterUtils.createStream(jssc, filters); + + JavaDStream<String> words = stream.flatMap(new FlatMapFunction<Status, String>() { + @Override + public Iterable<String> call(Status s) { + return Arrays.asList(s.getText().split(" ")); + } + }); + + JavaDStream<String> hashTags = words.filter(new Function<String, Boolean>() { + @Override + public Boolean call(String word) throws Exception { + return word.startsWith("#"); + } + }); + + // Read in the word-sentiment list and create a static RDD from it + String wordSentimentFilePath = "data/streaming/AFINN-111.txt"; + final JavaPairRDD<String, Double> wordSentiments = jssc.sparkContext().textFile(wordSentimentFilePath) + .mapToPair(new PairFunction<String, String, Double>(){ + @Override + public Tuple2<String, Double> call(String line) { + String[] columns = line.split("\t"); + return new Tuple2<String, Double>(columns[0], + Double.parseDouble(columns[1])); + } + }); + + JavaPairDStream<String, Integer> hashTagCount = hashTags.mapToPair( + new PairFunction<String, String, Integer>() { + @Override + public Tuple2<String, Integer> call(String s) { + // leave out the # character + return new Tuple2<String, Integer>(s.substring(1), 1); + } + }); + + JavaPairDStream<String, Integer> hashTagTotals = hashTagCount.reduceByKeyAndWindow( + new Function2<Integer, Integer, Integer>() { + @Override + public Integer call(Integer a, Integer b) { + return a + b; + } + }, new Duration(10000)); + + // 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 + JavaPairDStream<String, Tuple2<Double, Integer>> joinedTuples = + hashTagTotals.transformToPair(new Function<JavaPairRDD<String, Integer>, + JavaPairRDD<String, Tuple2<Double, Integer>>>() { + @Override + public JavaPairRDD<String, Tuple2<Double, Integer>> call(JavaPairRDD<String, + Integer> topicCount) + throws Exception { + return wordSentiments.join(topicCount); + } + }); + + JavaPairDStream<String, Double> topicHappiness = joinedTuples.mapToPair( + new PairFunction<Tuple2<String, Tuple2<Double, Integer>>, String, Double>() { + @Override + public Tuple2<String, Double> call(Tuple2<String, + Tuple2<Double, Integer>> topicAndTuplePair) throws Exception { + Tuple2<Double, Integer> happinessAndCount = topicAndTuplePair._2(); + return new Tuple2<String, Double>(topicAndTuplePair._1(), + happinessAndCount._1() * happinessAndCount._2()); + } + }); + + JavaPairDStream<Double, String> happinessTopicPairs = topicHappiness.mapToPair( + new PairFunction<Tuple2<String, Double>, Double, String>() { + @Override + public Tuple2<Double, String> call(Tuple2<String, Double> topicHappiness) + throws Exception { + return new Tuple2<Double, String>(topicHappiness._2(), + topicHappiness._1()); + } + }); + + JavaPairDStream<Double, String> happiest10 = happinessTopicPairs.transformToPair( + new Function<JavaPairRDD<Double, String>, JavaPairRDD<Double, String>>() { + @Override + public JavaPairRDD<Double, String> call(JavaPairRDD<Double, + String> happinessAndTopics) throws Exception { + return happinessAndTopics.sortByKey(false); + } + } + ); + + // Print hash tags with the most positive sentiment values + happiest10.foreachRDD(new Function<JavaPairRDD<Double, String>, Void>() { + @Override + public Void call(JavaPairRDD<Double, String> happinessTopicPairs) throws Exception { + List<Tuple2<Double, String>> topList = happinessTopicPairs.take(10); + System.out.println( + String.format("\nHappiest topics in last 10 seconds (%s total):", + happinessTopicPairs.count())); + for (Tuple2<Double, String> pair : topList) { + System.out.println( + String.format("%s (%s happiness)", pair._2(), pair._1())); + } + return null; + } + }); + + jssc.start(); + jssc.awaitTermination(); + } +} http://git-wip-us.apache.org/repos/asf/bahir/blob/c6155756/streaming-twitter/examples/src/main/scala/org/apache/spark/examples/streaming/twitter/TwitterHashTagJoinSentiments.scala ---------------------------------------------------------------------- diff --git a/streaming-twitter/examples/src/main/scala/org/apache/spark/examples/streaming/twitter/TwitterHashTagJoinSentiments.scala b/streaming-twitter/examples/src/main/scala/org/apache/spark/examples/streaming/twitter/TwitterHashTagJoinSentiments.scala new file mode 100644 index 0000000..0328fa8 --- /dev/null +++ b/streaming-twitter/examples/src/main/scala/org/apache/spark/examples/streaming/twitter/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