Spark学习--SparkStreaming

SparkStreaming

介绍

流式计算框架

批计算

数据已经存在, 一次性读取所有的数据进行批量处理

流计算

数据源源不断的进来, 经过处理后落地

特点

Spark Streaming 会源源不断的处理数据, 称之为流计算
Spark Streaming 并不是实时流, 而是按照时间切分小批量, 一个一个的小批量处理
Spark Streaming 是流计算, 所以可以理解为数据会源源不断的来, 需要长时间运行

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 状态, 准备数据发送

案例介绍

使用 Spark Streaming 程序和 Socket server 进行交互, 从 Server 处获取实时传输过来的字符串, 拆开单词并统计单词数量, 最后打印出来每一个小批次的单词数量

使用Netcat

源代码

package sparkStreaming

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}

/**
 * @author mooojl
 * @Date 2021-01-17-11:55
 */
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.47.100",
      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 的处理机制叫做小批量, 英文叫做 mini-batch, 是收集了一定时间的数据后生成 RDD, 后针对 RDD 进行各种转换操作, 这个原理提现在如下两个地方
- 控制台中打印的结果是一个批次一个批次的, 统计单词数量也是按照一个批次一个批次的统计
- 多长时间生成一个 RDD 去统计呢? 由 new StreamingContext(sparkConf, Seconds(1)) 这段代码中的第二个参数指定批次生成的时间
Spark Streaming 中至少要有两个线程

在使用 spark-submit 启动程序的时候, 不能指定一个线程
- 主线程被阻塞了, 等待程序运行
- 需要开启后台线程获取数据