|
一、SparkStreaming概述
1. 数据处理类型分类
- 静态数据
? - 数据源是不变的、有限的、显式离散的
? - 多适用于批量计算、离线计算
- 流数据
? - 数据是变动的、无限的、连续的
? - 多适用于实时计算，能在秒级、秒内处理完成
? ? - 实时数据分类
? ? ? - 小时级
? ? ? - 分钟级
? ? ? - 秒级
- 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
? ? ? ? - 前一个窗口到后一个窗口所经过的时间长度间隔。必须是批处理时间间隔的倍数
2. 处理流程图示说明
- 框架处理总流程图
- 框架内部工作流程图
? - Spark Streaming接收实时输入数据流并将数据分成批处理，然后由SparkCore引擎处理，以批量生成最终结果流。
二、scala快速构建sparkstreaming应用
1. 在maven中添加spark-streaming依赖
? ?```xml
? ? ? ? ? <!-- spark-streaming依赖 -->
? ? ? ? ?<dependency>
? ? ? ? ? ?<groupId>org.apache.spark</groupId>
? ? ? ? ? ?<artifactId>spark-streaming_${scala.compile.version}</artifactId>
? ? ? ? ? ?<version>2.3.2</version>
? ? ? ? ? ?<scope>provided</scope>
? ? ? ? </dependency>
? ?```
2. 以StreamingContext为起点，面向DStream编程
? ?```scala
? ?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()
? ? ?}
? ?}
? ?
? ?```
3. 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常见问题说明
1. 输入DStream和Receivers
- 输入DStream：即为从数据流源接受的输入数据流的DStream
? - 内置两类流媒体源
? ? - 基本流数据源：包括本地或是hdfs文件系统、Socket套接字链接、Akka actor等。
? ? - 高级流数据源：包括Kafka、Flume、Kinesis、ZeroMQ等数据源，可以通过引入第三方工具库来使用该数据源。
? - Receivers：每个输入DStream(除文件流之外)均与Receiver相对关联，该对象负责从流源接收数据并将其存储在Spark的内存中进行处理。
2. 关于SparkStreaming本地运行时线程数量的设置
- 在本地运行Spark Streaming程序时，不能使用“local”或“local [1]”作为主URL。
- 该设置即只有一个线程将用于本地运行任务。如果正使用基于接收器的输入DStream，则必须使用单个线程来运行接收器，而无法留下用于处理接收数据的线程。故本地运行时，始终使用“local [ n ]”作为主URL，其中n >要运行的接收器数量,即至少从2起
3. 关于SparkStreaming在集群运行时CPU逻辑核心数设置
- 一个逻辑CPU的资源相当于可以开启一个线程的能力。
- 在集群上运行时，分配给SparkStreaming应用程序的核心数必须大于接收器数。否则系统将只能接收数据，但无法处理数据。
|