SparkStreaming

1、SparkStreaming概述

  • 数据处理类型分类
    • 静态数据
      • 数据源是不变的、有限的、显式离散的
      • 多适用于批量计算、离线计算
    • 流数据
      • 数据是变动的、无限的、连续的
      • 多适用于实时计算，能在秒级、秒内处理完成
      • 实时数据分类
      • 小时级
      • 分钟级
      • 秒级
    • sparkstreaming是什么
      • 一句话总结：微批处理的流式(数据)实时计算框架。
      • 原理：是把输入数据以某一时间间隔批量的处理，当批处理间隔缩短到秒级时，即可用于处理实时数据流。
      • 优点
      • 可以和spark core、sparksql等无缝集成
      • 支持从多种数据源获取数据，包括Kafka、Flume、Twitter、ZeroMQ、Kinesis以及TCP sockets，然后可以使用诸如map、reduce、join等高级函数进行复杂算法的处理，最后可以将处理结果存储到HDFS文件系统，数据库等。
    • 重要概念说明
      • StreamingContext
      • 类比于SparkContext，SparkSqlContext
      • 流计算框架中的中枢类，负责各种环境信息、分发调度等任务。
      • 数据源
      • 简称：Source，意为DataSource的缩写
      • 指流数据的来源是哪里，如文件，Socket输入、Kafka等。
      • 离散流
      • 英文称Discretized Stream，简称DStream，即为sparkstreaming微批处理当中的数据抽象单位。
      • 是继spark core的RDD、spark sql的DataFrame和DataSet后又一基础的数据类型，是spark streaming特有的数据类型。
      • 输入离散流
      • 英文简称：Input DStream
      • 将Spark Streaming连接到一个外部Source数据源来读取数据的统称
      • 批数据
      • 英文称Batch Data
      • 连续数据离散化的步骤：将流式实时连续的数据整体转化成以时间片为单位进行分批，即将流式数据转化成时间片为单位数据进行批数据处理，随着时间推移，这些处理结果即形成结果数据流，即流处理引擎形成。
      • 时间片或批处理时间间隔
      • 英文称 batch interval
      • 人为对流数据进行定量的标准，以时间片作为拆分流数据的依据。
      • 一个时间片的数据对应一个RDD实例。
      • 窗口长度
      • 英文称window length
      • 一个窗口覆盖的流数据的时间长度，必须是批处理时间间隔的倍数。
      • 窗口分类
      • 滑动窗口
      • 滚动窗口
      • 滑动窗口时间间隔
      • 滑动窗口：简称Sliding window
      • 前一个窗口到后一个窗口所经过的时间长度间隔。必须是批处理时间间隔的倍数
  • 处理流程图示说明
    • 框架处理总流程图

5 SparkStreaming - 图1

  • 框架内部工作流程图

Spark Streaming接收实时输入数据流并将数据分成批处理，然后由SparkCore引擎处理，以批量生成最终结果流。
5 SparkStreaming - 图2

2、scala快速构建SparkStreaming应用

  • scala构建SparkStreaming应用步骤
    • sbt构建空eclipse项目(done)
    • build.sbt中添加spark-streaming依赖
    • 以StreamingContext为起点，而向Dstream编程
    • 测试运行、效果查看
  • 步骤实现
    • build.sbt中添加spark-streaming依赖

name := “FirstSparkStreaming4Sbt”
scalaVersion := “2.11.11”
autoScalaLibrary := false
organization := “com.tl.job002”
libraryDependencies ++= Seq(
//spark-core依赖 ,注意版本group对应通过%%实现
“org.apache.spark” %% “spark-core” % “2.3.2” % “provided”,
“org.apache.spark” %% “spark-sql” % “2.3.2” % “provided”,
“org.apache.spark” %% “spark-streaming” % “2.3.2” % “provided”
)

  • 以StreamingContext为起点，而向Dstream编程

package com.tl.job002.streaming
import org.apache.spark.
import org.apache.spark.streaming.
//SparkStreaming测试类
object SparkStreamingTest {
def main(args: Array[String]): Unit = {
// 要初始化Spark Streaming程序，必须创建一个StreamingContext对象，
//它是所有Spark Streaming功能的主要入口点。
//一切都从SparkConf开始
val conf = new SparkConf().setMaster(“local[2]”)
.setAppName(“NetworkWordCount”)
//指定时间间隔的ssc初始化
val ssc = new StreamingContext(conf, Seconds(1))

//ssc指定来自TCP源作为输入数据源，即链接一个指定主机的已打开的TCP端口，从该端口中读取文本数据，每行以”\n”作为每行的结尾。
val lines = ssc.socketTextStream(“localhost”, 9999)

//将DStream进行打平处理,实际是对其内部的离散的rdd进行打平处理
val words = lines.flatMap(.split(“\s+”))

// 将单列的word转化为双列的kv结构，用于后边的wc操作
val pairs = words.map(word => (word, 1))

//对kv的输出进行wc计算
val wordCounts = pairs.reduceByKey( + _)
//打印wc结果到控制台
wordCounts.print()

//正式开始计算
ssc.start()
//等待计算结束，一般流式计算没有异常或人为干预是一直保持运行状态的
ssc.awaitTermination()
}
}

  • 测试运行、效果查看
    • netcat介绍
      • 基本介绍
      • netcat是一个通过TCP/UDP在网络中进行读写数据工具（命令），被称为“瑞士军刀”，主要用于网络调试领域、传输领域甚至黑客攻击领域。
      • 简称nc，linux自带该命令， windows需要下载netcap的应用程序才行。
      • Linux版本
      • 启动netcat命令
      • nc -lk 9999 ：意为开启本机的9999端口作为对外的TCP数据服务端口
      • 开启本机的9999作为TCP服务端口

5 SparkStreaming - 图3

  1.   - windows版本
  2.   - 下载windows netcat
  3.   - [https://blog.csdn.net/qq_37585545/article/details/82250984](https://blog.csdn.net/qq_37585545/article/details/82250984)
  4.   - 解压下载完的压缩包后，找到nc64.exe所在的目录
  5.   - 启动数据输出命令
  6.   - ./nc64.exe -lp 9999

5 SparkStreaming - 图4

  • 开启sparkStreaming程序，读取来自9999端口的输入的TCP流数据
    • 注意要先开启指定端口的TCP服务，否则streaming会报读取异常。
    • 右击-run as application即可，开启后ss将一直以1秒为时间间隔，以行为输入单位监控传入的流数据，进行代码当中的flatMap、WordCount的处理
    • 每隔1秒，输出一次针对当前DStream的WC计算结果
  • SparkStreaming实现WordCount的流程抽象说明(经典流程)
    • SparkStreaming代码的开发流程
      • 初始化StreamingContext
      • 通过创建输入DStreams来定义输入源。
      • 通过将转换和输出操作应用于DStream来定义流式计算。
      • 开始接收数据并使用它进行处理streamingContext.start()。
      • 等待处理停止（手动或由于任何错误）使用streamingContext.awaitTermination()。
      • 可以使用手动停止处理streamingContext.stop()。
    • SparkStreaming代码开发注意
      • 一旦启动(start)了上下文，就不能设置或添加新的流式计算。
      • 上下文停止后，无法重新启动。
      • 在JVM中只能同时激活一个StreamingContext。
      • StreamingContext上的stop()也会停止SparkContext。要仅停止StreamingContext，请将stop()的可选参数设置stopSparkContext为false。
      • 只要在创建下一个StreamingContext之前停止前一个StreamingContext（不停止SparkContext），就可以重复使用SparkContext来创建多个StreamingContexts。
    • SparkStreaming的流数据专用数据抽象DStream，内部均为指定时间间隔内生成数据的RDD形式。

5 SparkStreaming - 图5

  • DStream内部即为RDD持续的离散序列，RDD内部均是以文本行为基本元素，对DStream的处理即映射转化为对内部RDD的处理，故对RDD的算子操作大多数情况下适配于DStream使用，同时DStream提供了更高级别的算子API方便使用。5 SparkStreaming - 图6

    3、SparkStreaming常见问题说明

  • 输入DStream和Receivers

    • 输入DStreams ：即为从数据流源接收的输入数据流的DStream。
      • 内置两类流媒体源
      • 基本流数据源：包括本地或是hdfs文件系统、Socket套接字链接、Akka actor等。
      • 高级流数据源：包括Kafka、Flume、Kinesis、ZeroMQ等数据源，可以通过引入第三方工具库来使用该数据源。
    • Receivers:每个输入DStream(除文件流之外)均与Receiver相对关联，该对象负责从流源接收数据并将其存储在Spark的内存中进行处理。
  • 关于多路输入流的处理
    • SparkStreaming对多路输入流进行自然和谐的支持。
    • 通过ssc可以同时创建多路输入流，并直接提供api进行各种的join、leftOuterJoin、rightOuterJoin等操作。
  • DStreams的输出操作
    • print():在运行流应用程序的驱动程序节点上打印DStream中每批数据的前十个元素。
    • saveAsTextFiles:将此DStream的内容保存为文本文件
    • saveAsObjectFiles:将此DStream的内容保存为SequenceFiles序列化Java对象。
    • saveAsHadoopFiles:将此DStream的内容保存为Hadoop文件。
    • foreachRDD:最通用的输出运算符，它将函数func应用于从流生成的每个RDD。此函数应将每个RDD中的数据推送到外部系统，例如将RDD保存到文件，或通过网络将其写入数据库。请注意，函数func在运行流应用程序的驱动程序Driver进程中执行。
    • foreach和foreachPartition运行在Worker节点。
  • 关于SparkStreaming本地运行时线程数量的设置
    • 在本地运行Spark Streaming程序时，不能使用“local”或“local [1]”作为主URL。
    • 该设置即只有一个线程将用于本地运行任务。如果正使用基于接收器的输入DStream，则必须使用单个线程来运行接收器，而无法留下用于处理接收数据的线程。故本地运行时，始终使用“local [ n ]”作为主URL，其中n >要运行的接收器数量,即至少从2起。

  • 关于SparkStreaming在集群运行时CPU逻辑核心数设置

    • 一个逻辑CPU的资源相当于可以开启一个线程的能力。
    • 在集群上运行时，分配给SparkStreaming应用程序的核心数必须大于接收器数。否则系统将只能接收数据，但无法处理数据。

      4、SparkStreaming处理高级数据源实战

  • 分布式消息中间件-Kafka应用实战专题篇

    5、经典问题

  • 以本地文件或目录和hdfs文件或目录为输入数据源，做实时数据流处理，实时记录每隔1s处理的数据行的行数。