<commit_before>package com.houseofmoran.spark.play.twitter

import java.io.FileReader
import java.util.Properties
import scala.collection.JavaConversions._

import org.apache.spark._
import org.apache.spark.streaming._
import StreamingContext._
import org.apache.spark.streaming.twitter._

object TwitterStream {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("TwitterStream").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(10))

    val oauthProperties = new Properties()
    oauthProperties.load(new FileReader(args(0)))
    for (key <- oauthProperties.stringPropertyNames()
         if key.startsWith("twitter4j.oauth.")) {
         System.setProperty(key, oauthProperties.getProperty(key))
    }

    val twitterStream = TwitterUtils.createStream(ssc, None)

    val wordStream = twitterStream.flatMap(status => status.getText().split(" ").map((_, 1)))
    val wordCountStream = wordStream.reduceByKey(_ + _)

    wordCountStream.print()

    ssc.start()
    ssc.awaitTermination()
  }
}<commit_msg>show word counts, over 1 minute chunks, grouped by frequency<commit_after>package com.houseofmoran.spark.play.twitter

import java.io.FileReader
import java.util.Properties
import scala.collection.JavaConversions._

import org.apache.spark._
import org.apache.spark.streaming._
import StreamingContext._
import org.apache.spark.streaming.twitter._

object TwitterStream {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("TwitterStream").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(1))

    val oauthProperties = new Properties()
    oauthProperties.load(new FileReader(args(0)))
    for (key <- oauthProperties.stringPropertyNames()
         if key.startsWith("twitter4j.oauth.")) {
         System.setProperty(key, oauthProperties.getProperty(key))
    }

    val twitterStream = TwitterUtils.createStream(ssc, None)

    val wordStream = twitterStream.flatMap(status => status.getText().split(" ").map((_, 1)))
    val wordCountStream = wordStream.reduceByKeyAndWindow((a: Int, b:Int) => a + b, Seconds(60), Seconds(60))

    wordCountStream.foreachRDD( wordCountRDD => {
        for(wordGroup <- wordCountRDD.groupBy{ case (word, count) => count }.sortBy(_._1)) {
          val count = wordGroup._1
          val words = wordGroup._2.map(_._1).mkString(", ")
          System.out.println(s"$count: $words")
        }
    });

    ssc.start()
    ssc.awaitTermination()
  }
}
{'fraction_non_alphanumeric': 0.1297923322683706, 'fraction_numerical': 0.00718849840255591, 'mean_word_length': 4.040241448692153, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1891775', 'n_tokens_mistral': 794, 'n_tokens_neox': 759, 'n_words': 151}