SparkStreaming

//测试Spark实时计算
object StreamWordCount {
  def main(args: Array[String]): Unit = {

    //创建配置对象
    val conf: SparkConf = new SparkConf().setAppName("Streaming").setMaster("local[*]")
    val streamContext: StreamingContext = new StreamingContext(conf, Seconds(5))
    //通过监控端口创建DStream，读进来的数据为一行行
    val socket: ReceiverInputDStream[String] = streamContext.socketTextStream("hadoop101", 9999)
    //将每一行数据做切分，形成一个个单词  读取是按一行一行来读 line ==> word
    val dsTream: DStream[String] = socket.flatMap(_.split(" "))
    //将单词映射成元组（word,1）
    val word: DStream[(String, Int)] = dsTream.map((_, 1))
    //reduceByKey
    val wordCount: DStream[(String, Int)] = word.reduceByKey(_+_)
      //打印
    wordCount.print()
    //启动采集器
    streamContext.start()
    //Driver不能停止，等待采集器的结束
    streamContext.awaitTermination()
  }

[kris@hadoop101 ~]$ nc -lk 9999
Hello world
Hello 
Hello java
Hello spark

// 自定义数据采集器
class CustomerReceive(host: String, port: Int) extends Receiver[String](StorageLevel.MEMORY_ONLY){ //有一个构造方法
  var socket: Socket = null
  //读数据并将数据发送给Spark
  def receive(): Unit = {
    //创建一个Socket
    val socket = new Socket(host, port)
    //字节流 ---->字符流
    val inputStream: InputStream = socket.getInputStream //字节流
    //字符流
    val bufferedReader: BufferedReader = new BufferedReader(new InputStreamReader(inputStream, "utf-8"))
    var line: String = null
    while ((line = bufferedReader.readLine()) != null){
      if (!"--END--".equals(line)){
        store(line) //存储到这里边
      }else{
        return
      }
    }
  }
  //启动采集器
  //最初启动的时候，调用该方法，作用为：读数据并将数据发送给Spark
  override def onStart(): Unit = {
    new Thread(new Runnable{
      override def run(): Unit = {
        receive()
      }
    }).start()
  }
  //关闭采集器
  override def onStop(): Unit = {
    if (socket != null){
      socket.close()
      socket = null
    }
  }

}

//测试：
object FileStream {
  def main(args: Array[String]): Unit = {
    // 创建流式处理环境对象
    // 创建对象时，需要传递采集数据的周期（时间）
    val conf: SparkConf = new SparkConf().setAppName("Streaming").setMaster("local[*]")
    val streamContext: StreamingContext = new StreamingContext(conf, Seconds(5))

    // 从端口号获取数据
    val socketDStream: ReceiverInputDStream[String] = streamContext.receiverStream(new CustomerReceive("hadoop101", 9999))
    // 一行一行的数据 line ==> word
    val wordDStream: DStream[String] = socketDStream.flatMap(_.split(" "))
    // word ==> (word, 1)
    val wordToCountDStream: DStream[(String, Int)] = wordDStream.map((_, 1))
    // reduceByKey
    val wordToSumDStream: DStream[(String, Int)] = wordToCountDStream.reduceByKey(_ + _)
    //打印数据
    wordToSumDStream.print()
    // TODO 启动采集器
    streamContext.start()
    // TODO Driver不能停止，等待采集器的结束
    // wait, sleep
    streamContext.awaitTermination()

  }
}

//监听kafka消息
object KafkaStreaming {
  def main(args: Array[String]): Unit = {
    // 创建配置对象
    val sparkConf = new SparkConf().setAppName("KafkaStreaming").setMaster("local[*]")
    // 创建流式处理环境对象
    // 创建对象时，需要传递采集数据的周期（时间）
    val socket: StreamingContext = new StreamingContext(sparkConf, Seconds(5))

    // 一个类如果创建SparkContext,那么这个类我们称之为Driver类
    // 从Kafka集群中获取数据
    //定义kafka参数
    val kafkaParams = Map[String, String](
      "group.id" -> "kris",
      "zookeeper.connect" -> "hadoop101:2181",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG ->"org.apache.kafka.common.serialization.StringDeserializer",//StringDeserializer的全类名，StringDeserializer implements Deserializer<String>
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer"
    ) //别导错包流，是kafka.clients.consumer里对
    //定义topic参数
    val topicMap = Map("thrid" -> 3)

    val kafkaDStream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream[String, String, StringDecoder, StringDecoder](
      socket,
      kafkaParams,
      topicMap,
      StorageLevel.MEMORY_ONLY) //StorageLevel别导错包流
    val wordToCountDStream = kafkaDStream.map {
      case (k, v) => {(v, 1)}
    }
    val wordToSumDStream: DStream[(String, Int)] = wordToCountDStream.reduceByKey(_ + _)
    //打印数据
    wordToSumDStream.print()
    //启动采集器
    socket.start()
    //Driver不能停，等待采集器对结束
    socket.awaitTermination()
  }
}

[kris@hadoop101 kafka]$ bin/kafka-console-producer.sh --broker-list hadoop101:9092 --topic thrid

//SparkStreaming有状态转换操作
object DStreamState {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("Stream").setMaster("local[*]")
    val streamContext: StreamingContext = new StreamingContext(conf, Seconds(5))
    //设置Checkpoints的目录
    streamContext.sparkContext.setCheckpointDir("cp")
    val socketDStream: ReceiverInputDStream[String] = streamContext.socketTextStream("hadoop101", 9999)

    val wordDStream: DStream[String] = socketDStream.flatMap(_.split(" "))

    val wordToCountDStream: DStream[(String, Int)] = wordDStream.map((_, 1))
//    进行有状态的转换操作
    val resultDStream: DStream[(String, Long)] = wordToCountDStream.updateStateByKey {// 要加范型
      case (seq, buffer) => {  //seq序列当前周期中单词对数量对集合， buffer表缓冲当中的值，所谓的checkPoint
        val sumCount = seq.sum + buffer.getOrElse(0L)
        Option(sumCount) //表往缓存里边更新对值　　它需要返回一个Option
      }
    }
    resultDStream.print()
    streamContext.start()
    streamContext.awaitTermination()

  }
}

object DStreamWindow {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Stream").setMaster("local[*]")
    val streamContext: StreamingContext = new StreamingContext(conf, Seconds(3))

    val socketDStream: ReceiverInputDStream[String] = streamContext.socketTextStream("hadoop101", 9999)

    // 设定数据窗口：window
    // 第一个参数表示窗口的大小（时间的范围，应该为采集周期的整数倍）
    // 第二个参数表示窗口的滑动的幅度（时间的范围，应该为采集周期的整数倍）
    val windowDStream: DStream[String] = socketDStream.window(Seconds(6), Seconds(3))
    val wordDStream: DStream[String] = windowDStream.flatMap(_.split(" "))
    val wordCountDStream: DStream[(String, Int)] = wordDStream.map((_, 1))
    val wordSumDStream: DStream[(String, Int)] = wordCountDStream.reduceByKey(_+_)

    wordSumDStream.print()
    streamContext.start()
    streamContext.awaitTermination()
  }
}

    // TODO XXXXXX (Drvier) * 1
    wordSumDStream.map{
      case(word, sum) => {
        // TODO YYYYYY (Executor) * N
        (word, 1)
      }
    }
    // transform可以将DStream包装好的RDD抽取出来进行转换操作
    // transform可以在每一个采集周期对rdd进行操作
    // TODO AAAAAA (Driver) * 1
    wordSumDStream.transform{
      rdd => {
        // TODO BBBBBBB (Driver) * N
        rdd.map{
          case (word, sum) => {
            // TODO CCCCCC (Executor) * N
            (word, 1)
          }
        }
      }
    }