%run "./025_c_extendedTwitterUtils2runWithLangs"

import org.apache.spark._

import org.apache.spark.storage._

import org.apache.spark.streaming._

import scala.math.Ordering

import twitter4j.auth.OAuthAuthorization

import twitter4j.conf.ConfigurationBuilder

%run "Users/raazesh.sainudiin@math.uu.se/scalable-data-science/secrets/026_secret_MyTwitterOAuthCredentials"

// ## Let's create a directory in dbfs for storing tweets in the cluster's distributed file system.

val outputDirectoryRoot = "/datasets/tweetsStreamTmp" // output directory

// to remove a pre-existing directory and start from scratch uncomment next line and evaluate this cell

dbutils.fs.rm(outputDirectoryRoot, true) 

// ## Capture tweets in every sliding window of slideInterval many milliseconds.

val slideInterval = new Duration(1 * 1000) // 1 * 1000 = 1000 milli-seconds = 1 sec

// Our goal is to take each RDD in the twitter DStream and write it as a json file in our dbfs.

// Create a Spark Streaming Context.

val ssc = new StreamingContext(sc, slideInterval)

// Create a Twitter Stream for the input source. 

val auth = Some(new OAuthAuthorization(new ConfigurationBuilder().build()))

val twitterStream = ExtendedTwitterUtils.createStream(ssc, auth)

// Let's import google's json library next.

import com.google.gson.Gson 

//Let's map the tweets into json formatted string (one tweet per line).

val twitterStreamJson = twitterStream.map(

                                            x => { val gson = new Gson();

                                                 val xJson = gson.toJson(x)

                                                 xJson

                                                 }

                                          ) 

val partitionsEachInterval = 1 

val batchInterval = 1 // in minutes

val timeoutJobLength =  batchInterval * 5

var newContextCreated = false

var numTweetsCollected = 0L // track number of tweets collected

twitterStreamJson.foreachRDD((rdd, time) => { // for each filtered RDD in the DStream

      val count = rdd.count()

      if (count > 0) {

        val outputRDD = rdd.repartition(partitionsEachInterval) // repartition as desired

        // to write to parquet directly in append mode in one directory per 'time'------------       

        val outputDF = outputRDD.toDF("tweetAsJsonString")

        // get some time fields from current `.Date()`

        val year = (new java.text.SimpleDateFormat("yyyy")).format(new java.util.Date())

        val month = (new java.text.SimpleDateFormat("MM")).format(new java.util.Date())

        val day = (new java.text.SimpleDateFormat("dd")).format(new java.util.Date())

        val hour = (new java.text.SimpleDateFormat("HH")).format(new java.util.Date())

        // write to a file with a clear time-based hierarchical directory structure for example

        outputDF.write.mode(SaveMode.Append)

                .parquet(outputDirectoryRoot+ "/"+ year + "/" + month + "/" + day + "/" + hour + "/" + time.milliseconds) 

        // end of writing as parquet file-------------------------------------

        numTweetsCollected += count // update with the latest count

      }

  })

// ## Let's start the spark streaming context we have created next.

ssc.start()

// total tweets downloaded

numTweetsCollected

// ## Go to SparkUI and see if a streaming job is already running. If so you need to terminate it before starting a new streaming job. Only one streaming job can be run on the DB CE.

// #  let's stop the streaming job next.

ssc.stop(stopSparkContext = false) 

StreamingContext.getActive.foreach { _.stop(stopSparkContext = false) } 

%run "./025_b_TTTDFfunctions"