DStreaming练习

package houpu.com.SparkStream

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

object SparkStreaming_Socket_Total {
  def main(args: Array[String]): Unit = {
    //currentValues:当前批次相同的单词出现的素有的1.（wangan,1）（wangan,1）（wangan,1）--->List(1,1,1)
    //historyValues:在之前所有批次中，相同单词出现的总次数
    def updateFunc(currentValues:Seq[Int],historyValues:Option[Int]):Option[Int]={
      val newValue:Int = currentValues.sum + historyValues.getOrElse(0)
      Some(newValue)
    }

    //创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setAppName("socket").setMaster("local[2]")
    //创建SparkContext
    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("warn")

    //创建StreamingContext需要一个SparkContext对象，还有一个批处理时间间隔，表示没5S处理一次数据
    val ssc:StreamingContext = new StreamingContext(sc, Seconds(5))

    //接受scoket数据
    val socketTextStream: ReceiverInputDStream[String] = ssc.socketTextStream("ip", 9999)


    //切分每一行数据
    val words:DStream[String] = socketTextStream.flatMap(_.split(" "))

    //每个单词记为1
    val pairs = words.map(word => (word, 1))
　　
　　//缓存rdd过程的数据

    ssc.checkpoint("hdfs://ip:9000/dict01/save/relt02")

    //向童单词出现累加1
    val result = pairs.updateStateByKey(updateFunc)

    //打印输出
    result.print()

    //启动并等待数据流的到来
    ssc.start()
    ssc.awaitTermination()

  }

}

package houpu.com.SparkStream

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

object SparkStreaming_SocketWindow {
  def main(args: Array[String]): Unit = {
    //创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setAppName("socket").setMaster("local[2]")
    //创建SparkContext
    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("warn")

    //创建StreamingContext需要一个SparkContext对象，还有一个批处理时间间隔，表示没5S处理一次数据
    val ssc:StreamingContext = new StreamingContext(sc, Seconds(5))

    //接受scoket数据
    val socketTextStream: ReceiverInputDStream[String] = ssc.socketTextStream("ip", 9999)

    //切分每一行数据
    val words:DStream[String] = socketTextStream.flatMap(_.split(" "))

    //每个单词记为1
    val pairs = words.map(word => (word, 1))

    //相同单词出现的1累加
    //reduceFunc:(V,V) => V，一个函数
    //windowOuration:Duration 窗口长度
    //slidDuration:Duration 滑动窗口的时间间隔，表示多久计算一次
    val result:DStream[(String,Int)] = pairs.reduceByKeyAndWindow((x:Int,y:Int) => x+y,Seconds(10),Seconds(10))

    //打印输出
    result.print()

    //启动并等待数据流的到来
    ssc.start()
    ssc.awaitTermination()

  }

}

 package houpu.com.SparkStream



import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}


object TransForm {
  def main(args: Array[String]) {

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

    //黑名单列表
    val blackList = ssc.sparkContext.parallelize(Array(("Mike",true),("Bob",true)))
    val socketTextStream: ReceiverInputDStream[String] = ssc.socketTextStream("master", 9999)

    val result = socketTextStream.map(line => {
      val userinfo = line.split(" ")
      (userinfo(0),userinfo(1))

    })
    //进行黑名单过滤，之筛选出不在黑名单
    result.transform(rdd =>{
      rdd.leftOuterJoin(blackList).filter(_._2._2.isEmpty)
    })print()

    //启动并等待数据流的到来
    ssc.start()
    ssc.awaitTermination()


  }
}
//leftOuterJoin类似于SQL中的左外关联left outer join，返回结果以前面的RDD为主，关联不上的记录为空。