package com.zhp.day12

import org.apache.spark._
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.receiver.Receiver

import java.lang.Thread.sleep // not necessary since Spark 1.3

object TestZidingyi extends App {

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

  private val value: ReceiverInputDStream[String] = ssc.receiverStream(new CustomReceiverssdd)
  value.print()
  ssc.start() // Start the computation
  ssc.awaitTermination() // Wait for the computation to terminate
}
//重点 自定义类 重写的  官网有相关案例
//https://spark.apache.org/docs/3.1.2/streaming-custom-receivers.html#implementing-a-custom-receiver
class CustomReceiverssdd
  extends Receiver[String](StorageLevel.MEMORY_AND_DISK_2) {

  def onStart() {
    // Start the thread that receives data over a connection
    new Thread("Socket Receiver") {
      override def run() {
        receive()
      }
    }.start()
  }

  def onStop() {

  }
  private def receive() {
    val strings = List(
      "sss",
      "sss",
      "sss",
      "sss",
      "sss",
      "sss"
    )
    for (elem <- strings) {
      store(elem)
      sleep(2000)
    }
  }
}