Spark学习--SparkStreaming

POOPE

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 启动程序的时候, 不能指定一个线程
  
  
  
  • 主线程被阻塞了, 等待程序运行
    
  • 需要开启后台线程获取数据