import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3

// Create a local StreamingContext with two working thread and batch interval of 1 second.
// The master requires 2 cores to prevent from a starvation scenario.

val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
val ssc = new StreamingContext(conf, Seconds(1))
// Seconds(1) : 1초 주기

// Create a DStream that will connect to hostname:port, like localhost:9999
val lines = ssc.socketTextStream("localhost", 9999)

// Split each line into words
val words = lines.flatMap(_.split(" "))

import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3
// Count each word in each batch
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)

// Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.print()

ssc.start()             // Start the computation
// start를 해야 시작!
ssc.awaitTermination()  // Wait for the computation to terminate

lazy val root = (project in file(".")).
  settings(
    inThisBuild(List(
      organization := "com.example",
      scalaVersion := "2.11.7",
      version      := "0.1.0-SNAPSHOT"
    )),
    name := "Hello",
    libraryDependencies ++= List(
      "org.twitter4j" % "twitter4j-core" % "4.0.6",
      "org.apache.bahir" % "spark-streaming-twitter_2.11" % "2.1.0",
      "org.apache.spark" % "spark-core_2.11" % "2.2.0",
      "org.apache.spark" % "spark-streaming_2.11" % "2.2.0"
      ),
    retrieveManaged := true
  )

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.twitter.TwitterUtils


object twitterStreaming extends App{
  println("hello")

  val appName = "spark_course"
  val master = "local[*]"
  val conf = new SparkConf().setAppName(appName).setMaster(master)
  val ssc = new StreamingContext(conf, Seconds(10))

  val rootLogger = Logger.getRootLogger()
  rootLogger.setLevel(Level.ERROR)

  val consumerKey = "****"
  val consumerSecret = "****"
  val accessToken = "****"
  val accessTokenSecret = "****"

  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 stream = TwitterUtils.createStream(ssc, None)

//  stream.print()
//  val text = stream.map(status => status.getText) // 아래처럼 리팩토링
  val text = stream.map(_.getText)

  val hashTags = text.flatMap(_.split(" ")).filter(_.startsWith("#"))

//  val hashTagCounts = hashTags.map(h => (h, 1)).reduceByKey(_+_)
  val hashTagCounts = hashTags.map(h => (h, 1)).reduceByKeyAndWindow(_+_, Seconds(60)) // Window 사용 : 60초의 데이터를 모음

//  text.print()
//  hashTags.print()
//  hashTagCounts.print() // 순서대로 정렬은 안됨..! foreachRDD를 사용하자

  hashTagCounts.foreachRDD {
    rdd =>
      println("=====")
      rdd.sortBy(_._2, false).take(10).foreach(println)
  }

  ssc.start()
  ssc.awaitTermination()
}