//设置package包名称以及导入依赖的类
package org.iamShantanu101.spark.SentimentAnalyzer.utils

import twitter4j.auth.OAuthAuthorization
import twitter4j.conf.ConfigurationBuilder


object OAuthUtils {
  def bootstrapTwitterOAuth(): Some[OAuthAuthorization] = {
    System.setProperty("twitter4j.oauth.consumerKey", PropertiesLoader.consumerKey)
    System.setProperty("twitter4j.oauth.consumerSecret", PropertiesLoader.consumerSecret)
    System.setProperty("twitter4j.oauth.accessToken", PropertiesLoader.accessToken)
    System.setProperty("twitter4j.oauth.accessTokenSecret", PropertiesLoader.accessTokenSecret)

    val configurationBuilder = new ConfigurationBuilder()
    val oAuth = Some(new OAuthAuthorization(configurationBuilder.build()))

    oAuth
  }
} 

//设置package包名称以及导入依赖的类
package com.databricks.apps

package twitterClassifier {
  import org.apache.spark.SparkContext
  import org.apache.spark.sql.SparkSession

  object SparkSetup {
    val spark = SparkSession
      .builder
      .appName(getClass.getSimpleName.replace("$", ""))
      .getOrCreate()

    val sqlContext = spark.sqlContext

    val sc: SparkContext = spark.sparkContext
    // Suppress "WARN BlockManager: Block input-0-1478266015800 replicated to only 0 peer(s) instead of 1 peers" messages
    sc.setLogLevel("ERROR")
  }
}

package object twitterClassifier {
  import org.apache.spark.mllib.linalg.Vector
  import org.apache.spark.mllib.feature.HashingTF
  import twitter4j.auth.OAuthAuthorization
  import twitter4j.conf.ConfigurationBuilder

  val numFeatures = 1000
  val tf = new HashingTF(numFeatures)

  def maybeTwitterAuth: Some[OAuthAuthorization] = Some(new OAuthAuthorization(new ConfigurationBuilder().build))

  
  def featurize(s: String): Vector = tf.transform(s.sliding(2).toSeq)
} 

//设置package包名称以及导入依赖的类
package com.aluxian.tweeather.streaming

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.receiver.Receiver
import twitter4j.auth.{Authorization, OAuthAuthorization}
import twitter4j.conf.ConfigurationBuilder
import twitter4j.{FilterQuery, Status}


class TwitterInputDStream(@transient ssc: StreamingContext,
                          twitterAuth: Option[Authorization],
                          filterQuery: Option[FilterQuery],
                          storageLevel: StorageLevel
                         ) extends ReceiverInputDStream[Status](ssc) {

  private val authorization = twitterAuth.getOrElse(createOAuthAuthorization())

  private def createOAuthAuthorization(): Authorization = {
    new OAuthAuthorization(new ConfigurationBuilder().build())
  }

  override def getReceiver(): Receiver[Status] = {
    new TwitterReceiver(authorization, filterQuery, storageLevel)
  }

} 

//设置package包名称以及导入依赖的类
package io.gzet.timeseries

import com.google.gson.GsonBuilder
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.twitter.TwitterUtils
import org.apache.spark.streaming.{Minutes, StreamingContext}
import org.apache.spark.{Logging, SparkConf, SparkContext}
import twitter4j.auth.OAuthAuthorization
import twitter4j.conf.ConfigurationBuilder

import scala.util.Try

object TwitterStream extends SimpleConfig with Logging {

  def getTwitterStream(ssc: StreamingContext, filters: Seq[String] = Nil) = {
    val builder = new ConfigurationBuilder()
    builder.setOAuthConsumerKey(twitterApiKey)
    builder.setOAuthConsumerSecret(twitterApiSecret)
    builder.setOAuthAccessToken(twitterTokenKey)
    builder.setOAuthAccessTokenSecret(twitterTokenSecret)
    val configuration = builder.build()
    TwitterUtils.createStream(
      ssc,
      Some(new OAuthAuthorization(configuration)),
      filters,
      StorageLevel.MEMORY_ONLY
    )
  }

  def main(args: Array[String]) = {

    val sparkConf = new SparkConf().setAppName("Twitter Extractor")
    val sc = new SparkContext(sparkConf)
    val ssc = new StreamingContext(sc, Minutes(5))

    val twitterStream = getTwitterStream(ssc, args).mapPartitions({ it =>
      val gson = new GsonBuilder().create()
      it map { s =>
        Try(gson.toJson(s))
      }
    })

    twitterStream
      .filter(_.isSuccess)
      .map(_.get)
      .saveAsTextFiles("twitter")

    // Start streaming context
    ssc.start()
    ssc.awaitTermination()

  }

} 

//设置package包名称以及导入依赖的类
import com.databricks.spark.corenlp.functions._
import org.apache.spark._
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{current_timestamp, explode}
import org.apache.spark.streaming._
import org.apache.spark.streaming.twitter.TwitterUtils
import twitter4j.auth.OAuthAuthorization
import twitter4j.conf.ConfigurationBuilder

object TwitterSentiment {
  System.setProperty("spark.cassandra.connection.host", "127.0.0.1")

  def main(args:Array[String]): Unit = {
    val conf = new SparkConf().setAppName("TwitterSentiment").setMaster("local[*]")

    val spark = SparkSession.builder.config(conf).getOrCreate()

    val ssc = new StreamingContext(spark.sparkContext, Seconds(1))
    // Create a Twitter Stream for the input source.
    val auth = Some(new OAuthAuthorization(new ConfigurationBuilder().build()))
    val twitterStream = TwitterUtils.createStream(ssc, auth, Array("trump", "donald"))

    val tweets = twitterStream
      .filter(tweet => tweet.getLang.equals("en") || tweet.getLang.equals(""))
      .map(_.getText())
      .map(_.replaceAll("/[^A-Za-z0-9 ]/", ""))
      .map(_.replaceAll("/", ""))
      .map(_.replaceAll("RT.+?(?=\\s)\\s", ""))
      .map(_.replaceAll("https([^\\s]+).*", ""))

    tweets.foreachRDD(rdd => {
      import spark.implicits._
      val df = rdd.toDF()

      val output = df.select(cleanxml('value).as('doc))
        .select(explode(ssplit('doc)).as('sen))
        .select('sen, tokenize('sen).as('words), ner('sen).as('nerTags), sentiment('sen).as('sentiment))


      val formated = output
        .drop("words")
        .drop("nerTags")
        .withColumnRenamed("sen", "tweet")
        .withColumn("processed_time", current_timestamp())

      formated
        .write
        .mode("append")
        .format("org.apache.spark.sql.cassandra")
        .options(Map("table" -> "storm_doris", "keyspace" -> "sentiment"))
        .save()

    })

    ssc.start()

    // Let's await the stream to end - forever
    ssc.awaitTermination()
  }
} 

//设置package包名称以及导入依赖的类
package org.p7h.spark.sentiment.utils

import twitter4j.auth.OAuthAuthorization
import twitter4j.conf.ConfigurationBuilder


object OAuthUtils {
  def bootstrapTwitterOAuth(): Some[OAuthAuthorization] = {
    System.setProperty("twitter4j.oauth.consumerKey", PropertiesLoader.consumerKey)
    System.setProperty("twitter4j.oauth.consumerSecret", PropertiesLoader.consumerSecret)
    System.setProperty("twitter4j.oauth.accessToken", PropertiesLoader.accessToken)
    System.setProperty("twitter4j.oauth.accessTokenSecret", PropertiesLoader.accessTokenSecret)

    val configurationBuilder = new ConfigurationBuilder()
    val oAuth = Some(new OAuthAuthorization(configurationBuilder.build()))

    oAuth
  }
}