spark学习进度28（SparkStreaming）

SparkStreaming

 

Spark Streaming 的特点

特点	说明
Spark Streaming 是 Spark Core API 的扩展	Spark Streaming 具有类似 RDD 的 API, 易于使用, 并可和现有系统共用相似代码 一个非常重要的特点是, Spark Streaming 可以在流上使用基于 Spark 的机器学习和流计算, 是一个一站式的平台
Spark Streaming 具有很好的整合性	Spark Streaming 可以从 Kafka, Flume, TCP 等流和队列中获取数据 Spark Streaming 可以将处理过的数据写入文件系统, 常见数据库中
Spark Streaming 是微批次处理模型	微批次处理的方式不会有长时间运行的 Operator, 所以更易于容错设计 微批次模型能够避免运行过慢的服务, 实行推测执行

 Netcat 的使用

Step 1: Socket 回顾

• Socket 是 Java 中为了支持基于 TCP / UDP 协议的通信所提供的编程模型

• Socket 分为 Socket server 和 Socket client

  Socket server

  监听某个端口, 接收 Socket client 发过来的连接请求建立连接, 连接建立后可以向 Socket client 发送 TCP packet 交互 (被动)

  Socket client

  向某个端口发起连接, 并在连接建立后, 向 Socket server 发送 TCP packet 实现交互 (主动)

• TCP 三次握手建立连接

  Step 1

  Client 向 Server 发送 SYN(j), 进入 SYN_SEND 状态等待 Server 响应

  Step 2

  Server 收到 Client 的 SYN(j) 并发送确认包 ACK(j + 1), 同时自己也发送一个请求连接的 SYN(k) 给 Client, 进入 SYN_RECV 状态等待 Client 确认

  Step 3

  Client 收到 Server 的 ACK + SYN, 向 Server 发送连接确认 ACK(k + 1), 此时, Client 和 Server 都进入 ESTABLISHED 状态, 准备数据发送

Step 2: Netcat

• Netcat 简写 nc, 命令行中使用 nc 命令调用

• Netcat 是一个非常常见的 Socket 工具, 可以使用 nc 建立 Socket server 也可以建立 Socket client

  • nc -l 建立 Socket server, l 是 listen 监听的意思

  • nc host port 建立 Socket client, 并连接到某个 Socket server

  流的演示：

  package cn.itcast.streaming
  
  import org.apache.spark.SparkConf
  import org.apache.spark.storage.StorageLevel
  import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
  import org.apache.spark.streaming.{Seconds, StreamingContext}
  
  object StreamingWordCount {
  
    def main(args: Array[String]): Unit = {
      // 1. 初始化环境
      // 在 SparkCore 中的内存, 创建 SparkContext 的时候使用
      // 在创建 Streaming Context 的时候也要用到 conf, 说明 Spark Streaming 是基于 Spark Core 的
      // 在执行 master 的时候, 不能指定一个线程
      // 因为在 Streaming 运行的时候, 需要开一个新的线程来去一直监听数据的获取
      val sparkConf = new SparkConf().setAppName("streaming word count").setMaster("local[6]")
      // StreamingContext 其实就是 Spark Streaming 的入口
      // 相当于 SparkContext 是 Spark Core 的入口一样
      // 它们也都叫做 XXContext
      val ssc = new StreamingContext(sparkConf, Seconds(5))
  
      ssc.sparkContext.setLogLevel("WARN")//减少日志时候使用
  
      // socketTextStream 这个方法用于创建一个 DStream, 监听 Socket 输入, 当做文本来处理
      // sparkContext.textFile() 创建一个 rdd, 他们俩类似, 都是创建对应的数据集
      // RDD -> Spark Core     DStream -> Spark Streaming
      // DStream 可以理解为是一个流式的 RDD
      val lines: ReceiverInputDStream[String] = ssc.socketTextStream(
        hostname = "192.168.15.128",//这是eth0 的
        port = 9999,
        storageLevel = StorageLevel.MEMORY_AND_DISK_SER
      )
  
      // 2. 数据的处理
      //    1. 把句子拆为单词
      val words: DStream[String] = lines.flatMap(_.split(" "))
      //    2. 转换单词
      val tuples: DStream[(String, Int)] = words.map((_, 1))
      //    3. 词频reduce
      val counts: DStream[(String, Int)] = tuples.reduceByKey(_ + _)
  
      // 3. 展示和启动
      counts.print()//并不会导致整个流执行
  
      ssc.start()
  
      // main 方法执行完毕后整个程序就会退出, 所以需要阻塞主线程
      ssc.awaitTermination()
    }
  }

  

  

   

   

  Spark Streaming 的特点

  • Spark Streaming 会源源不断的处理数据, 称之为流计算

  • Spark Streaming 并不是实时流, 而是按照时间切分小批量, 一个一个的小批量处理

  • Spark Streaming 是流计算, 所以可以理解为数据会源源不断的来, 需要长时间运行

  Spark Streaming 是按照时间切分小批量

  • 如何小批量?

    Spark Streaming 中的编程模型叫做 DStream, 所有的 API 都从 DStream 开始, 其作用就类似于 RDD 之于 Spark Core

    

    可以理解为 DStream 是一个管道, 数据源源不断的从这个管道进去, 被处理, 再出去

    

    但是需要注意的是, DStream 并不是严格意义上的实时流, 事实上, DStream 并不处理数据, 而是处理 RDD

    

    以上, 可以整理出如下道理

    • Spark Streaming 是小批量处理数据, 并不是实时流

    • Spark Streaming 对数据的处理是按照时间切分为一个又一个小的 RDD, 然后针对 RDD 进行处理

   RDD 和 DStream 的 DAG

  如果是 RDD 的 WordCount, 代码大致如下

  val textRDD = sc.textFile(...)
  val splitRDD = textRDD.flatMap(_.split(" "))
  val tupleRDD = splitRDD.map((_, 1))
  val reduceRDD = tupleRDD.reduceByKey(_ + _)

  

  用图形表示如下

  

  同样, DStream 的代码大致如下

  val lines: DStream[String] = ssc.socketTextStream(...)
  val words: DStream[String] = lines.flatMap(_.split(" "))
  val wordCounts: DStream[(String, Int)] = words.map(x => (x, 1)).reduceByKey(_ + _)

  同理, DStream 也可以形成 DAG 如下

  

  看起来 DStream 和 RDD 好像哟, 确实如此

  RDD 和 DStream 的区别

  

  • DStream 的数据是不断进入的, RDD 是针对一个数据的操作

  • 像 RDD 一样, DStream 也有不同的子类, 通过不同的算子生成

  • 一个 DStream 代表一个数据集, 其中包含了针对于上一个数据的操作

  • DStream 根据时间切片, 划分为多个 RDD, 针对 DStream 的计算函数, 会作用于每一个 DStream 中的 RDD

  DStream 如何形式 DAG

  

  • 每个 DStream 都有一个关联的 DStreamGraph 对象

  • DStreamGraph 负责表示 DStream 之间的的依赖关系和运行步骤

  • DStreamGraph 中会单独记录 InputDStream 和 OutputDStream