Movle

import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.storage.StorageLevel

object FlumeLogPull {
  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "/Users/macbook/Documents/hadoop/hadoop-2.8.4")
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    val conf = new SparkConf().setAppName("FlumeLogPull").setMaster("local[2]")
    val ssc = new StreamingContext(conf,Seconds(1))

    val flumeEvent = FlumeUtils.createPollingStream(ssc, "192.168.31.211", 1234,StorageLevel.MEMORY_ONLY_SER)

    val lineDStream = flumeEvent.map( e => {
      new String(e.event.getBody.array)
    })

    lineDStream.print()

    ssc.start()
    ssc.awaitTermination()
  }
}

import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.flume.FlumeUtils

object MyFlumeStream {

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

    System.setProperty("hadoop.home.dir", "/Users/macbook/Documents/hadoop/hadoop-2.8.4")
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    val conf = new SparkConf().setAppName("MyFlumeStream").setMaster("local[2]")

    val ssc = new StreamingContext(conf,Seconds(3))

    //创建 flume event 从 flume中接收push来的数据 ---> 也是DStream
    //flume将数据push到了 ip 和 端口中
    val flumeEventDstream = FlumeUtils.createStream(ssc, "192.168.1.121", 1236)

    val lineDStream = flumeEventDstream.map( e => {
      new String(e.event.getBody.array)
    })

    lineDStream.print()

    ssc.start()
    ssc.awaitTermination()
  }
}

import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import scala.collection.mutable.Queue
import org.apache.spark.rdd.RDD

object RDDQueueStream {

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

    System.setProperty("hadoop.home.dir", "/Users/macbook/Documents/hadoop/hadoop-2.8.4")
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    val conf = new SparkConf().setAppName("RDDQueueStream").setMaster("local[2]")

    val ssc = new StreamingContext(conf,Seconds(1))

    //需要一个RDD队列
    val rddQueue = new Queue[RDD[Int]]()


    for( i <- 1 to 3){
      rddQueue += ssc.sparkContext.makeRDD(1 to 10)

      Thread.sleep(5000)
    }

    //从队列中接收数据 创建DStream
    val inputDStream = ssc.queueStream(rddQueue)

    val result = inputDStream.map(x=>(x,x*2))

    result.print()

    ssc.start()
    ssc.awaitTermination()

  }
}

package day1211

import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.storage.StorageLevel

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

    System.setProperty("hadoop.home.dir", "/Users/macbook/Documents/hadoop/hadoop-2.8.4")
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    //local[2]代表开启两个线程
    val conf = new SparkConf().setAppName("MyNetwordWordCount").setMaster("local[2]")

    //接收两个参数，第一个conf，第二个是采样时间间隔
    val ssc = new StreamingContext(conf, Seconds(3))

    //监控目录 如果文件系统发生变化 就读取进来
    val lines = ssc.textFileStream("/Users/macbook/TestInfo/test_file_stream")

    lines.print()

    ssc.start()
    ssc.awaitTermination()
  }
}

import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.storage.StorageLevel
import org.apache.spark.sql.SparkSession

object MyNetwordWordCountWithSQL {

  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "/Users/macbook/Documents/hadoop/hadoop-2.8.4")
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    val conf = new SparkConf().setMaster("local[2]").setAppName("MyNetwordWordCountWithSQL")

    val ssc = new StreamingContext(conf,Seconds(5))

    val lines = ssc.socketTextStream("192.168.1.121",1235,StorageLevel.MEMORY_ONLY)

    val words = lines.flatMap(_.split(" "))

    //集成Spark SQL 使用SQL语句进行WordCount
    words.foreachRDD( rdd => {

      val spark = SparkSession.builder().config(rdd.sparkContext.getConf).getOrCreate()

      import spark.implicits._
      val df1 = rdd.toDF("word")

      df1.createOrReplaceTempView("words")

      spark.sql("select word , count(1) from words group by word").show()
    })

    ssc.start()
    ssc.awaitTermination()
  }

}

package day1211

import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.storage.StorageLevel

object MyTotalNetworkWordCount {

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

    System.setProperty("hadoop.home.dir", "/Users/macbook/Documents/hadoop/hadoop-2.8.4")
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    val conf = new SparkConf().setMaster("local[2]").setAppName("MyTotalNetworkWordCount")

    val ssc = new StreamingContext(conf,Seconds(3))

    //设置检查点目录，保存之前的状态信息
    ssc.checkpoint("hdfs://192.168.1.121:9000/TestFile/chkp0826")

    val lines = ssc.socketTextStream("192.168.1.121", 1235, StorageLevel.MEMORY_ONLY)

    val words = lines.flatMap(_.split(" "))

    val wordPair = words.map((_,1))

    /**
     * 两个参数：
     * 第一个参数：当前的值是多少
     * 第二个参数：之前的结果是多少
     */
    val addFunc = (curreValues:Seq[Int],previousValues:Option[Int]) => {
      //进行累加运算
      // 1、把当前值的序列进行累加
      val currentTotal = curreValues.sum

      //2、在之前的值上再累加

      Some( currentTotal + previousValues.getOrElse(0) )

    }

    //进行累加运算
    val total = wordPair.updateStateByKey(addFunc)

    total.print()

    ssc.start()
    ssc.awaitTermination()

  }
}

package day1211

import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.storage.StorageLevel

/**
 * 需求：每10秒，把过去30秒的数据取出来
 *
 * 窗口长度：30秒
 *
 * 滑动距离：10秒
 *
 */
object MyNetWorkWordCountByWindow {

  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "/Users/macbook/Documents/hadoop/hadoop-2.8.4")
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    val conf = new SparkConf().setAppName("MyNetWorkWordCountByWindow").setMaster("local[2]")

    val ssc = new StreamingContext(conf,Seconds(1))

    val lines = ssc.socketTextStream("192.168.1.121", 1235, StorageLevel.MEMORY_ONLY)

    val words = lines.flatMap(_.split(" ")).map((_,1))

    /**
     * reduce By Key And Window
     * 三个参数
     * 1、要进行什么操作
     * 2、窗口的大小
     * 3、窗口滑动的距离
     */

    val result = words.reduceByKeyAndWindow((x:Int,y:Int)=>(x+y),Seconds(30),Seconds(10))

    result.print()

    ssc.start()
    ssc.awaitTermination()
  }

}

package day1211
import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.storage.StorageLevel

/**
 * 知识点汇总：
 * 1、创建StreamingContext-->本质，核心：创建DStream
 *
 * 2、DStream的表现形式：就是一个RDD
 * 	操作DSteam和操作RDD是一样的。
 *
 * 3、使用DStream把连续的数据流编程不连续的RDD
 */
object MyNetwordWordCount {

  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "/Users/macbook/Documents/hadoop/hadoop-2.8.4")
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    //local[2]代表开启两个线程
    val conf = new SparkConf().setAppName("MyNetwordWordCount").setMaster("local[2]")

    //接收两个参数，第一个conf，第二个是采样时间间隔
    val ssc = new StreamingContext(conf,Seconds(3))

    //创建DStream 从netcat服务器上接收数据 因为接收字符串，所以使用textStream
    val lines = ssc.socketTextStream("192.168.1.121", 1234, StorageLevel.MEMORY_ONLY)

    val words = lines.flatMap(_.split(" "))

    val wordCount = words.map((_,1)).reduceByKey(_+_)
    //    val wordCount = words.transform(x => x.map((_,1))).reduceByKey(_+_)

    wordCount.print()

    ssc.start()

    //等待任务结束
    ssc.awaitTermination()

  }

}