[lxl@hadoop102 spark]$ sudo yum install nc.x86_64

[lxl@hadoop102 spark]$ nc -lk 9999

package com.atlxl.helloworld

import java.io.{BufferedReader, InputStreamReader}
import java.net.Socket
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.receiver.Receiver

/*
自定义输入流
 */

class CustomerRecevicer(host:String, port:Int) extends Receiver[String](StorageLevel.MEMORY_ONLY){

  //接收器启动的时候调用
  override def onStart(): Unit = {

    new Thread("receiver"){
      override def run(): Unit = {
        //接受数据并提交给框架
        receive()
      }
    }.start()

  }

  def receive(): Unit ={

    var socket: Socket = null
    var input: String = null

    try {
      socket = new Socket(host,port)

      //生成数据流
      val reader = new BufferedReader(new InputStreamReader(socket.getInputStream))

      //接收数据
      /*
      方式一：
       */
      while (!isStopped() && (input = reader.readLine()) != null){
        store(input)
      }
      /*
      方式二：
       */
//      input = reader.readLine()
//      store(input)
//      while (!isStopped() && input != null){
//        store(input)
//        input = reader.readLine()
//      }
//
      restart("restart")

    }catch {
      case e:java.net.ConnectException => restart("restart")
      case t:Throwable => restart("restart")
    }
  }

  //接收器关闭的时候调用
  override def onStop(): Unit = {}
}

package com.atlxl.helloworld

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

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

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

    //保存状态信息
    ssc.checkpoint("./check")

    //获取数据
//    val lineDStream = ssc.socketTextStream("hadoop102",9999)

    //自定义获取
    val lineDStream = ssc.receiverStream(new CustomerRecevicer("hadoop102", 9999))

    //DStream[String]
    val wordsDStream = lineDStream.flatMap(_.split(" "))

    //DStream[(String,1)]
    val k2vDStream = wordsDStream.map((_,1))

    //DStream[(String,sum)]
    //无状态转换
//    val result = k2vDStream.reduceByKey(_+_)

    val updateFuc =(v:Seq[Int],state:Option[Int])=> {

      val preStatus = state.getOrElse(0)
      Some(preStatus + v.sum)
    }

    //有状态转换
    val result = k2vDStream.updateStateByKey(updateFuc)
    result.print()

    //运行
    ssc.start()
    ssc.awaitTermination()

  }

}

val ipDStream = accessLogsDStream.map(logEntry => (logEntry.getIpAddress(), 1))
val ipCountDStream = ipDStream.reduceByKeyAndWindow(
 {(x, y) => x + y},
 {(x, y) => x - y},
 Seconds(30),
 Seconds(10))
 //加上新进入窗口的批次中的元素 //移除离开窗口的老批次中的元素 //窗口时长// 滑动步长

val dataset: RDD[String, String] = ...
val windowedStream = stream.window(Seconds(20))...
val joinedStream = windowedStream.transform { rdd => rdd.join(dataset) }