Spark3.x指北——4：Spark Streaming
Spark指北最后一部分内容：Spark Streaming~
虽然SparkStreaming不是目前Spark流处理的常用组件，但是其微批次转换为流处理的思路与Structured Streaming是相似的，可以大致了解一下。
Spark3.x指北全系列目录：
Spark基础概念情况：Spark3.x指北——1：Spark基础概念
SparkCore内容请看：Spark3.x指北——2：Spark Core
SparkSQL内容请看：Spark3.x指北——3：SparkSQL
目录
7.4.2 WordCount案例解析（SparkStreaming的工作流程）
⑤ reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])
c 测试是否能够通过SparkStreaming消费Kafka数据
a 连接Kafka，消费本次需求topic的数据，同时将数据封装到模型实体中
b 获取黑名单用户，对当前批次数据进行过滤，对过滤后的数据进行（key，1）的转换
c 对上一步过滤后的DStream进行wordcount聚合
① insertIntoBlackList——将用户 插入 / 更新 到黑名单table
② insertOrUpdateUserAdCount——将聚合结果 插入 / 更新 到用户点击table
③ getCount——从数据库中获取当前条件下用户的点击次数结果
① 让方法使用传入的connection参数，避免每个方法都要创建 & 关闭
② 优化connection的创建，避免对RDD每一条数据都要创建一次connection
a 消费Kafka数据，将其读取到SparkStreaming中，并将每条数据都封装到对应数据模型中
b 对数据进行map转换，转换为(key, 1)的形式，便于后续聚合统计
7 SparkStreaming
7.1 SparkStreaming概念 & 原理解析
SparkStreaming是基于SparkCore的，准实时、微批次的数据处理框架。
首先我们先来区分一下 实时/离线数据处理 & 流式/批次数据处理：
- 实时 & 离线（数据处理时间长短）：
- 实时：延迟通常以 毫秒 为单位
- 离线：延迟通常以 小时 / 天 为单位
- 流式 & 批次 （数据处理的方式）：
- 流式：来一条处理一条数据
- 批次：一次处理一批数据
然后我们就可以解析一下SparkStreaming的处理到底是什么样的了：
- 关于数据处理方式——微批次：
由于SparkStreaming的内核仍然是基于SparkCore，也就是说需要一台Driver与若干台Executor来进行任务处理，因此若一次仅仅处理一条数据则会造成其余Executor的闲置，浪费资源。所以，SparkStreaming处理的是一批数据，以充分利用Executor资源，因此其实际上是批次数据处理，而非真正的流式数据处理。
- 关于数据处理延迟——准实时：
同时，SparkStreaming想要做到准实时处理，就必须要求处理的一批数据不能太多，否则数据处理延迟就会重新达到sparkCore & sparkSQL的级别，因此SparkStreaming引入了微批次的概念，即将数据流按照小时、分钟等更小的单位进行切分，使得每次批处理的数据量不会过大，来保证其数据处理延迟能够达到 秒级延迟 ，同时又能充分运用SparkCore提供的资源。
Spark Streaming 支持的数据输入源很多，例如：Kafka、 Flume、Twitter、ZeroMQ 和简单的 TCP 套接字等等。数据输入后可以用 Spark 的高度抽象原语 如：map、reduce、join、window 等进行运算。而结果也能保存在很多地方，如HDFS，数据库等。
和Spark 基于RDD 的概念很相似，Spark Streaming 使用离散化流(discretized stream)作为抽象表示，叫作DStream。DStream 是随时间推移而收到的数据的序列。在内部，每个时间区间收到的数据都作为 RDD 存在，而DStream是由这些RDD所组成的序列(因此得名“离散化”)。所以 简单来将，DStream就是对RDD在实时数据处理场景的一种封装。
SparkStreaming的工作原理如下图所示：
结合我们介绍的准实时、微批次概念，SparkStreaming会将数据流按照更小的时间单位（如3s）划分为多个微批数据，由采集器将这些数据转化为DStream（一系列离散RDD），Driver会基于这些DStream划分stage、job，分发给Executor去做实际的取数据、计算数据的工作。
7.2 SparkStreaming特点
- 易用：
- 容错：
- 易整合：
7.3 SparkStreaming架构
7.3.1 结构图
- 整体架构：
- SparkStreaming架构：
7.3.2 背压机制
我们设想一个场景：很多时候，接收器接收速率过快会导致Driver端无法一次性处理那么多数据；而接收器接收速率过慢则会导致数据积压在数据流中。因此，调整接收器接收速率，来更高效运用SparkCore是很重要的。这就是背压机制的提出背景。
Spark 1.5 以前版本，用户如果要限制Receiver的数据接收速率，可以通过设置静态配制参 数“spark.streaming.receiver.maxRate”的值来实现，此举虽然可以通过限制接收速率，来适配当前 的处理能力，防止内存溢出，但也会引入其它问题。比如：producer数据生产高于maxRate，当 前集群处理能力也高于maxRate，这就会造成资源利用率下降等问题。
为了更好的协调数据接收速率与资源处理能力，1.5版本开始Spark Streaming可以动态控制 数据接收速率来适配集群数据处理能力。背压机制（即Spark Streaming Backpressure）: 根据 JobScheduler 反馈作业的执行信息来动态调整Receiver数据接收率。
通过属性“spark.streaming.backpressure.enabled”来控制是否启用 backpressure 机制，默认值 false，即不启用。
7.4 SparkStreaming入门
7.4.1 WordCount案例实操
需求：使用netcat工具向9999端口不断的发送数据，通过SparkStreaming读取端口数据并 统计不同单词出现的次数。
（1）添加依赖
<!--SparkStreaming依赖-->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.12</artifactId>
<version>3.0.0</version>
</dependency>再次注意，该依赖需要适配Scala & SparkCore的版本！
（2）代码实现
//SparkStreaming入门案例
object WordCount {
def main(args: Array[String]): Unit = {
//TODO 创建环境对象
//StreamingContext创建时，需要传递两个参数
//第一个参数：环境配置
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
//第二个参数：批处理的周期（采集周期 —— 一次采集采集多长周期内的数据）
val duration = Seconds(3)
//基于两个参数创建StreamingContext
val ssc = new StreamingContext(sparkConf, duration)
//TODO 逻辑处理
//获取端口数据，按行字符串获取
val lines = ssc.socketTextStream("localhost", 9999)
//WordCount操作
val words = lines.flatMap(_.split(" "))
val wordToOne = words.map((_, 1))
val wordToCount = wordToOne.reduceByKey(_ + _)
//结果输出
wordToCount.print()
//TODO 保持环境对象长期打开
//由于SparkStreaming采集器是长期执行任务，所以不能直接关闭
//如果main方法执行完毕，程序也会自动结束，因此main方法不能执行完毕
//1、启动采集器
ssc.start()
//2、等待采集器关闭，除非手动停止 / 超出运行时间
ssc.awaitTermination()
}
}几个注意事项：
- 就像SparkCore有自己的上下文SparkContext、SparkSQL有自己的上下文SparkSession一样，SparkStreaming组件也有自己的上下文StreamingContext。该context的创建需要两个参数，一个是常规的SparkConf配置对象，另一个则是批处理周期对象（即采集周期，简单理解就是每一批采集多少时间内的数据）。
- SparkStreaming需要不断接收采集器的数据，来进行流数据处理，因此不能像SparkCore/SparkSQL这类批处理计算一样，在计算结束后就调用stop()方法关闭上下文对象。StreamingContext需要使用start()方法 & awaitTermination()方法来保持采集器的开启，以源源不断接收数据。
（3）效果展示
- 首先，通过 nc -lp 9999（windows cmd命令）来向系统申请占用9999端口：
- 然后我们启动SparkStreaming应用程序，等待端口发送数据给采集器：
可以发现，控制台两条日志输出的时间间隔刚好为3ms。
- 在控制台9999端口发送数据，查看IDEA控制台流计算结果：
7.4.2 WordCount案例解析（SparkStreaming的工作流程）
Discretized Stream是Spark Streaming的基础抽象，代表持续性的数据流和经过各种Spark原语操作后的结果数据流。在内部实现上，DStream是一系列连续的RDD来表示。每个RDD含有 一段时间间隔内的数据。
简单来说，一个SparkStreaming数据可以理解为多个DStream（看这个流数据需要被几个业务场景应用），一个DStream内部按采集周期分为多个RDD，每个RDD包含一个采集周期内的数据。
应用于 DStream 的任何操作都会转换为对底层 RDD 的操作。例如，在前面的示例中，将行流转换为单词，flatMap 操作应用于 lines DStream 中的每个 RDD，以生成 words DStream 的 RDD。如下图所示：
最终的计算，由Spark计算引擎执行：
7.5 DStream创建
我们在7.4.2中提到，DStream是SparkStreaming中一个业务的流数据集合抽象，包含该流中所有批次的RDD数据，所以我们想要操作SparkStreaming，则首先需要创建DStream。
DStream的创建主要分为三种类型：
- 将具有周期性的RDD队列直接创建为DStream（测试用）
- 使用自定义数据源创建DStream（需要我们自己实现采集器，让SparkStreaming环境对象能够解析 & 采集，抽象为DStream）
- 将Kafka中生产者数据依据采集周期创建为DStream（企业中最常用的）
我们接下来分别对其进行介绍。
7.5.1 基于RDD队列创建DStream
测试过程中，可以通过使用ssc.queueStream(queueOfRDDs)来创建DStream，每一个推送到 这个队列中的RDD，都会作为一个DStream处理。
object DStreamWithQueue {
def main(args: Array[String]): Unit = {
//TODO 创建环境对象
//StreamingContext创建时，需要传递两个参数
//第一个参数：环境配置
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
//第二个参数：批处理的周期（采集周期 —— 一次采集采集多长周期内的数据）
val duration = Seconds(3)
//基于两个参数创建StreamingContext
val ssc = new StreamingContext(sparkConf, duration)
//TODO 从Queue中获取RDD作为DStream
val rddQueue = new mutable.Queue[RDD[Int]]()
val inputStream = ssc.queueStream(rddQueue, oneAtATime = false)
val result = inputStream
.map((_, 1))
.reduceByKey(_ + _)
result.print()
//TODO 保持环境对象长期打开
//由于SparkStreaming采集器是长期执行任务，所以不能直接关闭
//如果main方法执行完毕，程序也会自动结束，因此main方法不能执行完毕
//1、启动采集器
ssc.start()
//开启采集器后，不断向queue中添加RDD
for (i <- 1 to 5) {
rddQueue += ssc.sparkContext.makeRDD(1 to 300, 10) //创建RDD并加入队列中
Thread.sleep(1000) //等1000ms再运行下一个，给处理留出时间
}
//2、等待采集器关闭，除非手动停止 / 超出运行时间
ssc.awaitTermination()
}
}注意，对Queue的元素添加需要在采集器开始后进行，这样才能将队列中信息被采集器所接收。
7.5.2 基于自定义数据源创建DStream
StreamingContext需要基于采集器进行数据采集，以创建DStream。因此，我们基于自定义数据源创建DStream时，就需要自顶实现一个采集器。
我们首先需要通过继承Receiver，实现onStart & onStop方法，来自定义一个数据接收器，指定我们需要采集的数据，以及采集数据的逻辑（数据来源 & 数据接收逻辑）。
在实现了这个自定义数据采集器后，就可以通过 StreamingContext.receiverStream(自定义采集器) 方法，将自定义采集的数据转化为SparkStreaming可以处理的DStream格式，然后进行数据计算了。
以下是一个自定义数据源的示例：
object DStreamWithDIYDataSource {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("DStream")
val duration = Seconds(3)
val ssc = new StreamingContext(sparkConf, duration)
//使用自定义采集器
val messageDS = ssc.receiverStream(new MyReceiver)
messageDS.print()
//开启采集器
ssc.start()
//等待采集器关闭
ssc.awaitTermination()
}
/*
自定义数据采集器
1、定义泛型：接收数据的类型；传递参数：存储级别，此处指定为仅存储在内存
2、重写方法
*/
class MyReceiver extends Receiver[String](StorageLevel.MEMORY_ONLY_SER) {
//采集器是否执行的标识，默认为true，表示执行；当调用onStop方法会将其设置为false，即停止采集器
private var flag: Boolean = true
//启动数据采集过程
override def onStart(): Unit = {
//定义一个线程，用于生成自定义数据
new Thread(new Runnable {
override def run(): Unit = {
while (flag) {
//随机生产字符串，即消息
val message = "采集的数据为：" + new Random().nextInt(10).toString
//将消息按照存储级别进行封装
store(message)
Thread.sleep(500)
}
}
}).start()
}
//停止采集数据，清理释放资源
override def onStop(): Unit = {
flag = false
}
}
}（1）自定义数据源说明
我们需要通过继承Receiver来创建一个自定义数据源的采集器：
/*
自定义数据采集器
1、定义泛型：接收数据的类型；传递参数：存储级别，此处指定为仅存储在内存
2、重写方法
*/
class MyReceiver extends Receiver[String](StorageLevel.MEMORY_ONLY_SER) {
//采集器是否执行的标识，默认为true，表示执行；当调用onStop方法会将其设置为false，即停止采集器
private var flag: Boolean = true
//启动数据采集过程
override def onStart(): Unit = {
//定义一个线程，用于生成自定义数据
new Thread(new Runnable {
override def run(): Unit = {
while (flag) {
//随机生产字符串，即消息
val message = "采集的数据为：" + new Random().nextInt(10).toString
//将消息按照存储级别进行封装
store(message)
Thread.sleep(500)
}
}
}).start()
}
//停止采集数据，清理释放资源
override def onStop(): Unit = {
flag = false
}
}a 对于Recevier的继承，需要指定泛型 & 参数
-
- 泛型为：Receiver接收并存储的数据类型。
- 参数为：接收数据的存储级别，详细内容参考5.2.11-持久化。
b 实现onStart方法，用于指定数据如何被采集：
在该onStart方法中，我们通过创建一个新线程去创建数据，并且将数据进行采集。数据采集是通过 store()方法 将生成数据进行封装，该方法是SparkStreaming提供的。
数据采集的逻辑为 while循环条件 为true。为了与onStop方法逻辑对应，我们为这个自定义Receiver设置了Boolean类型的属性flag，用于采集是否进行的标识。flag = true时，表示需要生成 & 采集数据；flag = false时，表示停止生成 & 采集数据。
c 实现onStop方法，停止数据采集
当调用了采集器的onStop方法后，就会将线程循环执行的属性修改为false，阻止线程进一步生成 & 采集数据。
（2）基于这个自定义数据采集器的数据创建DStream
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("DStream")
val duration = Seconds(3)
val ssc = new StreamingContext(sparkConf, duration)
//使用自定义采集器
val messageDS = ssc.receiverStream(new MyReceiver)
messageDS.print()
//开启采集器
ssc.start()
//等待采集器关闭
ssc.awaitTermination()
}使用自定义接收器的核心是StreamingContext.receiverStream方法，该方法可以将我们自定义的数据采集器对象中的数据转化为SparkStreaming可处理的DStream格式，由Spark计算引起进行数据处理。
7.5.3 Socket数据的Receiver源码分析
我们在 7.5.2 实现的自定义数据采集是基于线程内随机生成数据实现的，较为简单。现在，我们基于 7.5.1 的网络wordcount，来查看一下Socket数据是如何实现接收的。
（1）SocketReceiver继承 & 属性分析
首先，Socket本质上也是一种自定义数据源，因此同样需要自定义接收器，使其可以被SparkStreaming接收并分析。所以，SocketReceiver同样实现了Receiver，同时拓展了该类所需要的参数（比如主机名、连接端口名等）：
该类同时混入了Logging特质，用于直接记录日志，排查连接错误。
其次，由于该类是用于接收Socket数据，所以设置了一个Socket对象作为该类的内部属性，以便在同一个SocketReceiver中直接进行对属性的修改：
（2）SocketReceiver核心方法分析
SocketReceiver内部主要有三个核心方法，除了 用于开启接收器并指定接收逻辑的onStart 以及 用于关闭接收器并且清理资源的onStop方法，还有一个建立socket连接，接收来自socket数据的方法receive。
a onStart方法
该方法其实核心与我们自定义实现的接收器很像，都是新建一个线程来进行数据接收（这同时也是自定义接收器接收数据的常用处理手段），通过这个手段可以避免阻塞SparkStreaming任务的核心线程。
同时，为了使用socket，在onStart方法中进行了socket的创建，基于传入的主机名 & 端口创建对应的socket，用于在接收数据的线程中使用（具体在receive方法内）。
b onStop方法
在onStop方法中，就是进行资源清理和接收器的关闭了。为了避免Socket被多个线程同时访问，在关闭时需要对Socket资源上锁，然后再进行关闭 & 置空 清理。
c receive方法
receive方法是onStart调用的，用于定义接收数据逻辑的方法。
该方法的核心逻辑其实很简单，创建了字节对象流，将socket传入的输入流转化为iterator，然后遍历iterator，将内部数据通过store进行封装。
由于Socket可能因为某些原因被关闭，所以通过finally代码块，在Socket被关闭后执行finally方法，强制调用onStop方法关闭这个接收器，来确保资源的回收（连接都没了等也没有了）。
7.5.4 Kafka数据源（重点）
ReceiverAPI：需要一个专门的Executor去接收数据，然后发送给其他的Executor做计算。存在 的问题，接收数据的Executor和计算的Executor速度会有所不同，特别在接收数据的Executor 速度大于计算的Executor速度，会导致计算数据的节点内存溢出。早期版本中提供此方式，当 前版本不适用。
DirectAPI：是由计算的Executor来主动消费Kafka的数据，速度由自身控制。
由于ReceiverAPI & Kafka 0-8 Direct的DirectAPI当前SparkStreaming版本都不适用，因此此处仅介绍 Kafka 0-10 Direct模式的数据源读取。
（1）需求
通过SparkStreaming从Kafka读取数据，并将读取过来的数据做简单计算，最终打印 到控制台。
（2）导入依赖
<!--sparkstreaming - Kafka相关依赖-->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
<version>2.10.1</version>
</dependency>（3）创建Kafka topic
[root@node1 kafka]# bin/kafka-topics.sh --bootstrap-server node1:9092 --create --replication-factor 3 --partitions 3 --topic muo我们创建 topic muo，让SparkStreaming从该topic中消费流数据。
（4）编写代码
object DStreamWithKafka {
def main(args: Array[String]): Unit = {
//TODO 准备环境
val sparkConf = new SparkConf().setAppName("DStreamWithKafka").setMaster("local[*]")
val duration = Seconds(3)
val ssc = new StreamingContext(sparkConf, duration)
//TODO 从Kafka中读取数据
//定义Kafka配置参数
val kafkaPara: Map[String, Object] = Map[String, Object](
//消费者服务器节点
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092,node2:9092,node3:9092",
//消费者组消费主题
ConsumerConfig.GROUP_ID_CONFIG -> "muo",
"key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)
//连接Kafka，获取消费数据
val kafkaDataDS = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferBrokers, //采集数据位置的策略，让框架自行匹配如何采集数据
ConsumerStrategies.Subscribe[String, String](Set("muo"), kafkaPara) //设置消费者相关参数
)
//查看kafka的数据
kafkaDataDS.map(_.value()).print()
//TODO 开启SparkStreaming接收器
ssc.start()
ssc.awaitTermination()
}
}a 设置Kafka配置参数
//定义Kafka配置参数
val kafkaPara: Map[String, Object] = Map[String, Object](
//消费者服务器节点
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092,node2:9092,node3:9092",
//消费者组消费主题
ConsumerConfig.GROUP_ID_CONFIG -> "muo",
"key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)我们需要指定Kafka的一些配置参数Map，比如kafka的broker节点所在地址、消费者组的主题等。
b 通过KafkaUtils连接Kafka，获取主题数据
//连接Kafka，获取消费数据
val kafkaDataDS = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferBrokers, //采集数据位置的策略，让框架自行匹配如何采集数据
ConsumerStrategies.Subscribe[String, String](Set("muo"), kafkaPara) //设置消费者相关参数
)通过SparkStreaming消费Kafka数据，首先肯定需要将StreamingContext作为参数传入。
其次，需要指定获取消费数据的策略。一般来说，都是让框架自己找合适的数据，因此使用LocationStrategies.PreferBrokers作为参数。
最后就是指定消费者的订阅参数了，即：指定topic & 指定Kafka连接参数。
c 查看Kafka数据
//查看kafka的数据
kafkaDataDS.map(_.value()).print()这是我们通过createDirectStream获取的Kafka流数据，其返回值为(k, v)对，v为topic中的元素，因此我们通过KafkaDataDS.map(_.value)来获取topic的元素，然后通过DStream的print方法打印元素 & 时间戳即可：
（5）查看消费情况
首先，在kafka集群中，通过命令创建一个生产者：
[root@node2 kafka]# bin/kafka-console-producer.sh --bootstrap-server node1:9092 --topic muo然后就可以输入消息，在IDEA控制台查看输出内容了：
7.6 DStream转换
DStream 上的操作与RDD的类似，分为Transformations（转换）和Output Operations（输 出）两种，此外转换操作中还有一些比较特殊的原语，如：updateStateByKey()、transform()以及 各种Window相关的原语。
7.6.1 状态操作说明
DStream的转换分为无状态操作 & 有状态操作。接下来我们先对二者进行区分。
（1）无状态操作演示
无状态操作，即仅针对当前批次数据进行的操作，不依赖当前DStream中的其余历史数据。
我们通过WordCount例子来解释一下无状态这个概念：
object wordcount {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
val ssc = new StreamingContext(sparkConf, Seconds(3))
//连接端口，获取输入流数据
val lines = ssc.socketTextStream("localhost", 9999)
val results = lines.flatMap(_.split(" ").map((_, 1))).reduceByKey(_ + _)
results.print()
ssc.start()
ssc.awaitTermination()
}
}上述代码是一个无状态wordcount，仅统计当前批次的wordcount。其运行结果如下所示：
可以看到，我们通过一系列无状态操作（flatMap、reduceByKey等）处理流数据，其控制台实际上是对每一批数据进行处理，但是这一批数据的处理并不包含之前的统计。比如，第二批的wordcount仅统计第二批次的数据，并不会将第一批的数据一起重新统计。
（2）有状态操作演示
有状态操作，不仅仅需要处理当前批次的数据，同时也要对DStream中的其余历史处理结果进行处理（或与历史数据关联）。简单理解就是，当前批次数据需要依赖历史数据进行操作。
当我们需要不仅需要统计这一批次的数据，同时要将前面批次的数据一起统计（其实就相当于基于当前批次数据更新统计情况），我们就需要使用有状态的转换操作：
object wordcount {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
val ssc = new StreamingContext(sparkConf, Seconds(3))
//连接端口，获取输入流数据
val lines = ssc.socketTextStream("localhost", 9999)
//设置检查点
ssc.checkpoint("cp")
val wordToOne = lines.flatMap(_.split(" ").map((_, 1)))
//通过有状态操作，依据当前批次数据更新历史的统计结果
//updateStateByKey操作，需要传递两个参数
// 参数一：当前批次相同key的value数据集合
// 参数二：缓冲区内相同key的value数据（历史数据统计后的结果）
val results = wordToOne.updateStateByKey(
(values: Seq[Int], buffValues: Option[Int]) => {
//获取缓冲区历史结果 & 当前批次统计结果，更新为一个新的统计结果
val newCount = buffValues.getOrElse(0) + values.sum
//将新统计结果封装为Option类型返回（防止NullPointException）
Option(newCount)
}
)
results.print()
ssc.start()
ssc.awaitTermination()
}
}我们进行的有状态操作为updateStateByKey，即根据历史阶段进行当前批次数据的wordcount：
有状态操作会比较复杂，但也很容易理解。不同于无状态操作仅需要指定对当前数据进行的操作，有状态操作需要针对当前批次的数据，对历史数据进行更新。因此对于有状态操作参数的操作函数，就需要当前批次数据 & 历史数据参数。我们以代码中用到的updateStateByKey函数为例，分析一下内部参数构成：
与无状态操作相同，在使用该函数时都需要传入操作函数，但是该操作函数具有两个参数：
Seq[V]：相同key下，当前批次的流数据的value
Option[S]：相同key下，历史数据的处理结果
然后我们就可以基于这两个参数指定处理逻辑，最后封装为Option作为流处理的最新历史结果返回。
比如我们在代码中指定的操作为，基于当前批次value进行求和，获取当前批次的wordcount，然后与历史数据处理结果进行更新，将当前key的最新wordcount进行封装返回：
需要注意的是，有状态操作必须基于checkpoint运行，毕竟历史处理数据需要有个地方保存下来才行：
我们进行测试，查看一下有状态的wordcount：
不难发现，对于有状态wordcount，是会将历史数据与当前批次数据结合进行处理的。
7.6.2 无状态DStream转换
所谓的无状态，就是指仅针对当前处理批次的数据进行计算。部分无状态转化操作列在了下表中。注意，针对键值对的DStream转化操作(比如 reduceByKey())要添加 import StreamingContext._才能在 Scala 中使用。
|
函数名称 |
目的 |
Scala示例 |
函数签名 |
|
map() |
对 DStream 中的每个元素应用给定函数，返回由各元素输出的元素组成的 DStream。 |
ds.map(x => x + 1) |
f: (T) -> U |
|
flatMap() |
对 DStream 中的每个元素应用给定函数，返回由各元素输出的迭代器组成的 DStream。 |
ds.flatMap(x => x.split(" ")) |
f: T -> Iterable[U] |
|
filter() |
返回由给定 DStream 中通过筛选的元素组成的 DStream。 |
ds.filter(x => x != 1) |
f: T -> Boolean |
|
repartition() |
改变 DStream 的分区数。 |
ds.repartition(10) |
N/A |
|
reduceByKey() |
将每个批次中键相同的记录归约。 |
ds.reduceByKey((x, y) => x + y) |
f: (T, T) -> T |
|
groupByKey() |
将每个批次中的记录根据键分组。 |
ds.groupByKey() |
N/A |
需要记住的是，尽管这些函数看起来像作用在整个流上一样，但事实上每个DStream在内部是由许多RDD（批次）组成，且无状态转化操作是分别应用到每个RDD上的。
例如：reduceByKey()会归约每个时间区间中的数据，但不会归约不同区间之间的数据。
大部分无状态操作与SparkCore的RDD转换算子没有什么差别，因此接下来，我们会介绍一些较为不同的无状态操作。
（1）transform
transform允许 DStream 上执行任意的RDD-to-RDD函数。即使这些函数并没有在DStream 的API中暴露出来，通过该函数可以方便的扩展Spark API。该函数每一批次调度一次。其实也就是对DStream中的RDD应用转换。
这意味着，我们可以直接获取DStream底层的所有RDD，直接对这些RDD进行更复杂的操作。
object transform {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
val ssc = new StreamingContext(sparkConf, Seconds(3))
//连接端口，获取输入流数据
val lines = ssc.socketTextStream("localhost", 9999)
//transform方法可以获取底层RDD，然后进行操作
lines.transform(rdd1 => rdd2)
ssc.start()
ssc.awaitTermination()
}
}transform操作除了可以执行一些DStream无法实现的功能外，还能够实现周期性代码执行（Driver端）。比如我们看下面两个例子：
这两个代码都是想要对获取的lines批数据进行map操作。对于直接通过DStream.map操作，map操作执行是位于Executor端；而transform操作中，对批数据rdd的转换是在Driver端执行的，只有RDD.map操作才是在Executor端执行。
这带来的好处就是，我们可以在transform获取底层RDD时（也就是代码内部），在Driver端周期性的执行一些代码。简单来说，transform操作为我们额外提供了一个放置周期性执行Driver端代码的手段。
总结而言，若一些操作DStream可以胜任，完全没必要使用transform。只有当某些场合下DStream功能不全，或者想要周期性执行某些代码时，才会考虑使用transform。
（2）join
与RDD的join算子类似，都是按key进行连接。不过DStream的join操作是将两个DStream的数据按照批次进行按Key的连接。
注意，两个流之间的join需要两个流的批次大小一致，这样才能做到同时触发计算。计算过程就是 对当前批次的两个流中各自的RDD进行join，与两个RDD的join效果相同。
object join {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
val ssc = new StreamingContext(sparkConf, Seconds(5))
//连接端口，获取输入流数据
val line9999 = ssc.socketTextStream("localhost", 9999)
val line8888 = ssc.socketTextStream("localhost", 8888)
val map9999: DStream[(String, Int)] = line9999.map((_, 9))
val map8888: DStream[(String, Int)] = line8888.map((_, 8))
val joinDS = map8888.join(map9999)
joinDS.print()
ssc.start()
ssc.awaitTermination()
}
}其实DStream的join操作，底层就是对两个DStream底层的批次数据RDD的join操作：
7.6.3 有状态DStream转换
（1）updateStateByKey
UpdateStateByKey 原语用于记录历史记录，有时，我们需要在DStream中跨批次维护状态(例如流计算中累加wordcount)。针对这种情况，updateStateByKey()为我们提供了对一个状态变量 的访问，用于键值对形式的DStream。给定一个由(键，事件)对构成的 DStream，并传递一个指 定如何根据新的事件更新每个键对应状态的函数，它可以构建出一个新的 DStream，其内部数据为(键，状态)对。
updateStateByKey() 的结果会是一个新的DStream，其内部的RDD 序列是由每个时间区间对 应的(键，状态)对组成的。
updateStateByKey 操作使得我们可以在用新信息进行更新时保持任意的状态。为使用这个功能，需要做下面两步：
- 定义状态，状态可以是一个任意的数据类型。
- 定义状态更新函数，用此函数阐明如何使用之前的状态和来自输入流的新值对状态进行更新。
同时注意，使用updateStateByKey 需要对检查点目录进行配置，会使用检查点来保存状态。
object wordcount {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
val ssc = new StreamingContext(sparkConf, Seconds(3))
//连接端口，获取输入流数据
val lines = ssc.socketTextStream("localhost", 9999)
//设置检查点
ssc.checkpoint("cp")
val wordToOne = lines.flatMap(_.split(" ").map((_, 1)))
//通过有状态操作，依据当前批次数据更新历史的统计结果
//updateStateByKey操作，需要传递两个参数
// 参数一：当前批次相同key的value数据集合
// 参数二：缓冲区内相同key的value数据（历史数据统计后的结果）
val results = wordToOne.updateStateByKey(
(values: Seq[Int], buffValues: Option[Int]) => {
//获取缓冲区历史结果 & 当前批次统计结果，更新为一个新的统计结果
val newCount = buffValues.getOrElse(0) + values.sum
//将新统计结果封装为Option类型返回（防止NullPointException）
Option(newCount)
}
)
results.print()
ssc.start()
ssc.awaitTermination()
}
}代码解析部分在7.6.1-（2）中有进行说明，此处不再赘述。
（2）WindowOperations
a window操作概念解析
DStream中的window操作实际上与Scala中的sliding操作很类似，只不过Scala中以数据个数作为窗口大小 & 滑动步长的依据，而DStream中则以采集周期作为窗口大小 & 滑动步长依据。
下图是对window操作的一个解析。我们有一个SparkStreaming，内部的DStream以3s为一个采集周期，而我们的window是一个窗口大小为6s（2 * 采集周期）、步长为3s（1 * 采集周期）的窗口。即：该窗口每次会获取6s的批次数据，每次获取后会向后移动3s，接着获取下一批次的数据。
以文字来说，就是假设我们有9s的流数据，采集周期为3s（那么就有3个微批次数据），现在有一个6s窗口大小、3s滑动步长的窗口，那么滑动获取流数据过程即：
------------------------------------------------------------------------------
-
- 第一次滑动：还未移动到流数据中
- 第二次滑动：1s，2s，3s（批次①）
- 第三次滑动：1s，2s，3s（批次①），4s，5s，6s（批次②）
- 第四次滑动：4s，5s，6s（批次②），7s，8s，9s（批次③）
- 第五次滑动：7s，8s，9s（批次③）
- 第六次滑动：流数据处理结束
-------------------------------------------------------------------------------
需要注意：
- 与Scala的sliding操作不同，由于DStream数据是具有采集周期概念的，因此窗口相关参数都需要是采集周期的整数倍，否则可能会出现某个批次部分数据无法被完整采集的情况！
- 由于窗口需要从数据流入口慢慢向后滑动，也就代表着一开始可能不会获取到流数据，只有当窗口按照采集周期慢慢向后移动，才会慢慢接收、处理数据。因此，window操作对数据数量的获取应当依据 少 -> 多 -> 多 -> … -> 少 的曲线规律来进行。
- DStream窗口的滑动步长默认为 1 * 采集周期，就如上述例子中所示。这就导致了这样会出现某些批次数据可能被重复处理。为避免这种情况，可以显示设置窗口滑动步长。不过，并不是所有数据被重复处理都是需要避免的情况，这要根据实际的业务情景来考虑。
b 最基础的window操作——window
object window {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
val ssc = new StreamingContext(sparkConf, Seconds(3))
//连接端口，获取输入流数据
val lines = ssc.socketTextStream("localhost", 9999)
val wordToOne = lines.map((_, 1))
//windowDuration:
// 窗口的范围应该是采集周期的整数倍（n倍），即一个窗口可以获取n个批次的数据
//slideDuration:
// 窗口默认情况下，按照一个采集周期为步长进行滑动
// 比如采集周期为3，，窗口范围为6，共有9批数据，步长为3，每次获取的数据即：1, 2, 3, 4, 5, 6 -> 4, 5, 6, 7, 8, 9
// 很明显，会有被重复计算的数据
// 我们可以改变滑动的周期（即滑动步长），比如使得每一次滑动与窗口范围相同，这样就可以防止数据被重复计算
val windowsDS = wordToOne.window(Seconds(6), Seconds(6))
val wordToCount = windowsDS.reduceByKey(_ + _)
wordToCount.print()
ssc.start()
ssc.awaitTermination()
}
}DStream.window是对DStream数据按照窗口设置进行滑窗，将滑窗结果作为新的DStream返回。
由于其仅进行按照窗口划分数据，因此具有两个参数：
-
-
- windowDuration：窗口周期，也就是窗口大小。窗口每次会采集这个大小的批次数据作为新DStream返回。该参数一定一定一定要是采集周期的整数倍，防止批次内某些数据丢失！
- slideDuration：滑动周期，也就是滑动步长。窗口每次会按照这个步长进行滑动。根据业务场景将步长设置为采集周期的整数倍（默认为1 * 采集周期）。
-
c window操作的其余补充
除了基础的window对DStream按窗口进行划分外，还有一些其余操作，允许我们对每个窗口内的数据进行聚合。注意，这些数据 / 元素 是指DStream在该窗口内的RDD中的数据，不是指RDD本身！
部分操作如下：
① countByWindow
返回一个包含滑动窗口中元素的个数的DStream。
② countByValueAndWindow
返回一个包含滑动窗口中每个元素的个数的DStream。
与countByWindow不同之处在于，countByWindow不区分元素是否相同，是元素就算一个；countByValueAndWindow则会对元素进行区分，同种元素单独计算个数。
③ reduceByWindow
通过自定义聚合，对当前窗口中的流数据进行聚合，返回一个新的DStream。
④ reduceByKeyAndWindow
当在一个数据类型为(K,V) 的DStream上调用此函数，会对窗口内的value按key使用自定义函数处理，返回一个新(K,V)对的DStream。
⑤ reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])
这个函数是上述函数的变化版本，每个窗口的reduce值都是通过用前一个窗的reduce值来递增计算。
通过reduce进入到滑动窗口数据并”反向reduce”离开窗口的旧数据来实现这个操作。一个例子是随着窗口滑动对keys的“加”“减”计数。
简单理解就是在对新的窗口数据进行reduce的同时，会将已经计算过且仍然在窗口内的数据进行剔除，防止计算重复数据。
通过前边介绍可以想到，这个函数只适用于”可逆的reduce函数”，也就是这些reduce函数有相应的”反reduce”函数(以参数invFunc形式 传入)。
可逆函数在此处指的是能精准的移除离开窗口的贡献数据的系列函数。比如 加法 ，离开窗口的加和是很容易计算的，因此可以精准移除；而对于取 max 等函数，无法判断离开窗口贡献数据是否包含最大值，因此无法精准移除。
在SparkStreaming中的可逆函数说明如下：
- ✅ 加法、减法、计数：完全可逆
- ⚠️ 乘法、除法：有条件可逆（注意零值）
- ❌ 最大值、最小值、平均值：直接形式不可逆
- 🔧 复杂聚合：可以通过拆解为可逆组件来实现
如前述函数，reduce任务的数量通过可选参数来配置。
以下是使用reduceByKeyAndWindow对已经滑动走的数据进行剔除的例子：
object reduceByKeyAndWindow {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
val ssc = new StreamingContext(sparkConf, Seconds(3))
ssc.checkpoint("checkpoint") //有状态操作，设置检查点！
//连接端口，获取输入流数据
val lines = ssc.socketTextStream("localhost", 9999)
val wordToOne = lines.map((_, 1))
val windowDS = wordToOne.reduceByKeyAndWindow(
(x: Int, y: Int) => x + y, //聚合方式（window内数据如何聚合）
(x: Int, y: Int) => x - y, //剔除滑动窗口外已经处理过的数据的方式
Seconds(9), //窗口范围
Seconds(3) //滑动步长
)
windowDS.print()
ssc.start()
ssc.awaitTermination()
}
}7.7 DStream输出
输出操作指定了对流数据经转化操作得到的数据所要执行的操作(例如把结果推入外部数据库 或输出到屏幕上)。与RDD中的惰性求值类似，如果一个DStream及其派生出的DStream都没 有被执行输出操作，那么这些DStream就都不会被求值。如果StreamingContext中没有设定输出 操作，整个context就都不会启动。
输出操作如下：
- print()：在运行流程序的驱动结点上打印DStream中每一批次数据的最开始10个元素。这 用于开发和调试。在Python API中，同样的操作叫print()。注意，时间戳是print方法底层自带的，若其余输出操作需要时间戳等信息需要自行编写。
- saveAsTextFiles(prefix, [suffix])：以 text 文件形式存储这个 DStream 的内容。每一批次的存 储文件名基于参数中的prefix和suffix。”prefix-Time_IN_MS[.suffix]”。
- saveAsObjectFiles(prefix, [suffix])：以 Java 对象序列化的方式将 Stream 中的数据保存为 SequenceFiles . 每一批次的存储文件名基于参数中的为"prefix-TIME_IN_MS[.suffix]". Python 中目前不可用。
- saveAsHadoopFiles(prefix, [suffix])：将 Stream 中的数据保存为 Hadoop files. 每一批次的存 储文件名基于参数中的为"prefix-TIME_IN_MS[.suffix]"。Python API 中目前不可用。
- foreachRDD(func)：这是最通用的输出操作，即将函数 func 用于产生于 stream的每一个 RDD。其中参数传入的函数func应该实现将每一个RDD中数据推送到外部系统，如将 RDD存入文件或者通过网络将其写入数据库。
通用的输出操作foreachRDD()，它用来对DStream中的RDD运行任意计算。这和transform() 有些类似，都可以让我们访问任意RDD。在foreachRDD()中，可以重用我们在Spark中实现的 所有行动操作。比如，常见的用例之一是把数据写到诸如MySQL的外部数据库中。
这是一个foreachRDD输出操作的示例：
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
val ssc = new StreamingContext(sparkConf, Seconds(3))
val lines = ssc.socketTextStream("localhost", 9999)
val wordToOne = lines.flatMap(_.split(" ").map((_, 1)))
val windowDS = wordToOne.reduceByKeyAndWindow(
(x, y) => x + y,
(x, y) => x - y,
Seconds(9),
Seconds(3)
)
//该方法不会出现时间戳
windowDS.foreachRDD(println(_))
ssc.start()
ssc.awaitTermination()注意事项：
- 连接不能写在driver层面（序列化）
- 如果写在foreach则每个RDD中的每一条数据都创建，得不偿失；
- 增加foreachPartition，在分区创建（获取）。
7.8 优雅地关闭
7.8.1 如何进行优雅关闭
流式任务需要7*24小时执行，但是有时涉及到升级代码需要主动停止程序，但是分布式程序，没办法做到一个个进程去杀死，所有配置优雅的关闭就显得至关重要了。使用外部文件系统来控制内部程序关闭。
下面是一个简单的优雅关闭的示例：
object CloseGraceFully {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
val ssc = new StreamingContext(sparkConf, Seconds(3))
val lines = ssc.socketTextStream("localhost", 9999)
val wordToOne = lines.flatMap(_.split(" ").map((_, 1)))
val windowDS = wordToOne.reduceByKeyAndWindow(
(x, y) => x + y,
(x, y) => x - y,
Seconds(9),
Seconds(3)
)
windowDS.print()
//启动采集器
ssc.start()
//如果想要关闭采集器，那么需要创建新的线程
//而且需要在第三方程序中增加关闭状态，交给第三方控制
new Thread(
new Runnable {
override def run(): Unit = {
//优雅地关闭~
//计算节点不再接收新的数据，而是将现有的数据处理完毕，然后关闭
//一般来说，需要由第三方来决定是否关闭
//比如：
// MYSQL中有一张StopSpark的表，通过表中字段更新（yes or no）来告诉该线程关闭ssc
// Redis中使用K-V类型，通过更新V来告诉该线程关闭ssc
while (true) {
if (true) { //模拟：若MySQL无法读取到数据，那就关闭
val state = ssc.getState() //获取ssc的状态
if (state == StreamingContextState.ACTIVE) //活跃状态才关闭
ssc.stop(stopSparkContext = true, stopGracefully = true)
}
}
}
}
).start()
ssc.awaitTermination() //等待采集器关闭，会阻塞main线程
}
}由于ssc.awaitTermination会阻塞main线程，而ssc.start之前的代码采集器还未开启，所以对SparkStreaming的关闭需要放在这两个代码块之间。
优雅关闭的核心思路是new一个线程，在该线程中通过第三方来监控Spark的状态，当状态发生变更时在该线程中执行关闭方法。
当然，优雅关闭可以在外部创建一个函数来实现，可以有效减少代码耦合。
7.8.2 优雅关闭后数据恢复
在程序关闭后，若想要恢复上一次关闭时的数据然后继续进行计算，则需要通过checkpoint机制来实现。
这是一个简单的数据恢复代码示例：
object CloseGraceFully_DataResume {
def main(args: Array[String]): Unit = {
//设置数据保存 & 数据恢复
val ssc = StreamingContext.getActiveOrCreate(
"cp", //数据保存的检查点
() => { //如何创建SparkContext环境对象
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
val ssc = new StreamingContext(sparkConf, Seconds(3))
//配置checkpoint
ssc.checkpoint("cp")
val lines = ssc.socketTextStream("localhost", 9999)
val wordToOne = lines.flatMap(_.split(" ").map((_, 1)))
val windowDS = wordToOne.reduceByKeyAndWindow(
(x, y) => x + y,
(x, y) => x - y,
Seconds(9),
Seconds(3)
)
windowDS.print()
ssc
}
)
//启动采集器
ssc.start()
ssc.awaitTermination() //等待采集器关闭，会阻塞main线程
}
}核心在于StreamingContext.getActiveOrCreate这个API。该API会查看是否存在检查点，若存在则加载数据并返回其中的上下文对象，若不存在则直接创建新的上下文对象。
该API需要我们显示指定两个参数：
- checkpointPath：恢复数据的检查点来源，数据中包含上下文对象
- creatingFunc：若检查点不存在时，如何创建上下文对象
当checkpointPath中存在数据，那么就直接读取这个检查点中的数据，将检查点的上下文对象返回；若不存在时，则通过creatingFunc参数指定的创建方法来创建一个新的上下文对象。
我们示例代码中，将业务逻辑放到了创建方法中。这并不是唯一选择，业务逻辑也可以放置在外部，看生产环境的要求。
但需要注意，一定要在创建上下文对象中配置检查点（需要与getActiveOrCreate中的checkpointPath参数相同），这样才能让检查点与数据恢复协同工作！
7.9 SparkStreaming案例实操
7.9.1 环境准备
（1）案例所需要的maven依赖
<properties>
<maven.compiler.source>11</maven.compiler.source>
<maven.compiler.target>11</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<spark.version>3.0.0</spark.version>
<scala.version>2.12.10</scala.version>
</properties>
<dependencies>
<!-- Spark 核心依赖 -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>${spark.version}</version>
</dependency>
<!-- Spark SQL 依赖 -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>${spark.version}</version>
</dependency>
<!-- Spark Hive 支持（只需要这一个） -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.12</artifactId>
<version>${spark.version}</version>
</dependency>
<!-- MySQL JDBC -->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.33</version>
</dependency>
<!-- MySQL Connector -->
<dependency>
<groupId>com.mysql</groupId>
<artifactId>mysql-connector-j</artifactId>
<version>8.0.33</version>
</dependency>
<!--MySQL数据库连接池——采用Druid-->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid</artifactId>
<version>1.2.18</version>
</dependency>
<!--SparkStreaming依赖-->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<!-- 如果需要额外的 Hive 元数据支持，添加以下依赖 -->
<!--
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-metastore</artifactId>
<version>2.3.9</version>
</dependency>
-->
<!--sparkstreaming - Kafka相关依赖-->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<!--json解析类库-->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
<version>2.10.1</version>
</dependency>
</dependencies>（2）生成模拟数据源 & 测试流数据消费情况
由于SparkStreaming需要流数据源，也就是需要实时性的生成数据，因此无法像SparkCore & SparkSQL的案例实操（5.4 & 6.8）那样，通过现有的数据文件来进行批处理。所以我们需要借助Kafka，周期性生成模拟数据到Kafka中，然后从中消费数据作为数据源，对这一批次数据源进行分析。
a 创建Kafka topic
[root@node1 kafka]# bin/kafka-topics.sh --bootstrap-server node1:9092 --create --replication-factor 3 --partitions 3 --topic sparkstreamingpracticeb 创建生成模拟数据的Application
package main.spark_streaming.final_practice
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import java.util.Properties
import scala.collection.mutable.ListBuffer
import scala.util.Random
/**
* 案例流程：//Application => Kafka => SparkStreaming => 统计分析
* 该模块为Application，用于生成模拟数据，推送到Kafka，让SparkStreaming实时消费
*/
object MockData {
def main(args: Array[String]): Unit = {
//生成模拟数据
//数据格式： timestamp area city userid adid（空格隔开）
//数据含义： 时间戳 区域 城市 用户 广告
//创建一个Kafka生产者
val producer = createKafkaProducer("node1:9092")
//生产数据到Kafka
produceToKafka(producer)
}
/**
* 生成模拟数据，一批生成30条数据
* 生成的数据每一条都遵循格式：timestamp area city userid adid（空格隔开）
* @return
*/
def generatingMockData(): ListBuffer[String] = {
val mockDataList = ListBuffer[String]() //用于存储生成数据的集合
val areaList = List("华东", "华北", "华南", "华中") //area列表，用于随机数据的area参数
val cityList = List("北京", "上海", "深圳", "广州", "厦门") //city列表，用于随机数据的city参数
for (i <- 1 to 30) {
val area = areaList(new Random().nextInt(4)) //随机选一个area
val city = cityList(new Random().nextInt(5)) //随机选一个city
val userid = new Random().nextInt(6) //随机生成一个user
val adid = new Random().nextInt(6) //随机生成一个点击广告
mockDataList.append(
s"${System.currentTimeMillis()} ${area} ${city} ${userid} ${adid}"
)
}
mockDataList
}
/**
* 创建Kafka生成者
* @param broker Kafka leader服务器节点url
* @return
*/
def createKafkaProducer(broker: String): KafkaProducer[String, String] = {
// 创建配置对象
val prop = new Properties()
// 添加配置
prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker)
prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer")
prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer")
// 根据配置创建Kafka生产者
new KafkaProducer[String, String](prop)
}
/**
* 向Kafka推送数据的核心流程
* @param kafkaProducer Kafka生产者，用于推送数据
*/
def produceToKafka(kafkaProducer: KafkaProducer[String, String]): Unit = {
while (true) {
//调用生产随机数据，并将所有内容推送到Kafka中
generatingMockData().foreach(
data => {
println("生成的模拟数据为：" + data)
//创建一个生产者record，指定生产到的 topic & 数据内容，用于封装向Kafka发送的数据
val record = new ProducerRecord[String, String]("sparkstreamingpractice", data)
//向Kafka推送数据
kafkaProducer.send(record)
}
)
//每2s生产、发送一次即可
Thread.sleep(2000)
}
}
}该程序分为三个核心模块：
① 用于批量生成数据的模块
我们在该方法中批量生成30条模拟数据，然后将其封装为一个List进行返回。
注意：每一条模拟数据的格式均为：”timestamp area city userid adid” ，每个数据间用空格进行分隔。
② 用于创建Kafka生产者的模块
该模块主要用于生成一个Kafka生产者。我们需要通过Properties指定配置创建一个Kafka的生产者。
注意：由于Kafka只能通过leader节点进行读 / 写，因此该broker需要为leader节点的url。
③ 用于向Kafka推送数据的核心模块
这是向Kafka推送流式数据的核心模块。我们首先需要为该模块传入一个Kafka producer，然后在该模块中，我们通过while循环定期生成随机数据并且通过producer向Kafka中进行生产。
Kafka的数据生产分为两个步骤：
- 将要生产的数据封装为record
- 将record通过生产者推送到Kafka中
c 测试是否能够通过SparkStreaming消费Kafka数据
首先创建一个消费者，指定消费者配置，通过StreamingContext将Kafka作为数据源，生成流数据：
/**
* 测试从Kafka中消费数据是否成功
*/
object ConsumerTest {
def main(args: Array[String]): Unit = {
//TODO 准备环境
val sparkConf = new SparkConf().setAppName("DStreamWithKafka").setMaster("local[*]")
val duration = Seconds(3)
val ssc = new StreamingContext(sparkConf, duration)
，
//TODO 从Kafka中读取数据
//定义Kafka配置参数
val kafkaPara: Map[String, Object] = Map[String, Object](
//消费者服务器节点
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092,node2:9092,node3:9092",
//消费者组消费主题
ConsumerConfig.GROUP_ID_CONFIG -> "sparkstreamingpractice",
"key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)
//连接Kafka，获取消费数据
val kafkaDataDS = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferBrokers, //采集数据位置的策略，让框架自行匹配如何采集数据
ConsumerStrategies.Subscribe[String, String](Set("sparkstreamingpractice"), kafkaPara) //设置消费者相关参数
)
//查看kafka的数据
kafkaDataDS.map(_.value()).print()
//TODO 开启SparkStreaming接收器
ssc.start()
ssc.awaitTermination()
}
}启动该程序 与 b中的生产者程序，然后查看控制台是否有消费记录输出（注意先启动zookeeper & Kafka）：
7.9.2 需求一
（1）思路分析
该需求实际上并不复杂，只是我们需要通过多次校验，来保证能够完美检测到黑名单用户。
总体文字流程如下：
- 首先对于Kafka得到的微批次数据，若其中有部分数据本身就位于黑名单，则我们不需要对这数据进行任何操作。这样可以减少后续需要进行聚合的数据量，增加性能。注：图中map于黑名单校验顺序有误。
- 对校验后的数据，我们需要进行一个转换，将其转换为(key, 1)的形式，然后进行wordcount。对于这个需求，需要对同一用户、同一广告、同一天的点击进行统计，所以key = (用户信息, 广告信息, 当天时间戳)。
- 对于上述得到的数据，我们对其进行聚合，然后判断这一批次数据的聚合结果中，用户点击数是否超过了进入黑名单的阈值，若超过了则同样拉入到黑名单中；若未超过则更新 / 插入数据库。注意，这次的校验是针对这一批次数据进行校验，而非全部数据进行校验。若这一批数据不符合要求，可以不插入数据库，这同样可以提升性能。
- 对于更新后的数据库数据，还需要进行最后校验。若最终数据超过了阈值，同样要将该用户拉入黑名单，防止后续对这个用户不断进行数据库更新等无效操作。
黑名单总共有三次校验：
- 对于Kafka中原始数据进行校验，防止黑名单数据进入后续统计流程。
- 当前DStream批次数据统计结束，校验这一批数据是否有需要进入黑名单的，防止黑名单数据被插入数据库。
- 对于插入数据库后的最终数据进行校验，这次校验的结果将会提供给后续批次数据进行使用。
（2）一些提前准备
a MySQL数据库 & 表的创建
create database spark_streaming_practice;
use spark_streaming_practice;
# 用户黑名单表
create table black_list(
userid char(1) primary key
);
# 用户每日点击数据
create table user_ad_count(
dt varchar(255), #日期
userid char(1), #用户id
adid char(1), #广告id
count bigint, #点击次数统计
primary key (dt, userid, adid)
);我们需要两张表来实现这个需求：
- black_list：记录用户是否为黑名单用户的表
- user_ad_count：记录用户当天广告点击清空的表
b JDBCUtil工具类准备
package main.spark_streaming.final_practice.util
import com.alibaba.druid.pool.DruidDataSourceFactory
import java.sql.Connection
import java.util.Properties
import javax.sql.DataSource
/**
* JDBC相关操作的封装工具类
*/
object JDBCUtil {
var dataSource: DataSource = init()
/**
* 初始化数据库连接池
* @return
*/
private def init(): DataSource = {
val url = "jdbc:mysql://localhost:3306/spark_streaming_practice"
val driver = "com.mysql.jdbc.driver"
val user = "root"
val password = "D200504193010"
val properties = new Properties()
//设置JDBC的连接参数
properties.setProperty("url", url)
properties.setProperty("driver", driver)
properties.setProperty("user", user)
properties.setProperty("password", password)
properties.setProperty("maxActive", "50") //最大连接池数量
DruidDataSourceFactory.createDataSource(properties)
}
/**
* 获取MySQL连接
* @return
*/
def getConnection: Connection = {
dataSource.getConnection
}
}我们通过一个JDBCUtil来封装对MySQL连接的逻辑，除了该需求中使用外，对于其余需求也可以直接通过这个连接去连接到MySQL。
若出现Druid连接池配置读取乱码，可以直接在url中配置所有信息（user、password等），比如：
/**
* 初始化数据库连接池
* @return
*/
private def init(): DataSource = {
val url = "jdbc:mysql://localhost:3306/spark_streaming_practice?useUnicode=true&characterEncoding=UTF-8&useSSL=false&user=root&password=D200504193010"
val driver = "com.mysql.cj.jdbc.Driver"
val properties = new Properties()
//设置JDBC的连接参数
properties.setProperty("url", url)
properties.setProperty("driver", driver)
properties.setProperty("maxActive", "50") //最大连接池数量
properties.setProperty("connectionProperties", "useUnicode=true;characterEncoding=UTF-8")
DruidDataSourceFactory.createDataSource(properties)
}（3）代码实现
- 完整业务代码如下：
package main.spark_streaming.final_practice
import main.spark_streaming.final_practice.util.JDBCUtil
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import java.text.SimpleDateFormat
import java.util.Date
import scala.collection.mutable.ListBuffer
/**
* 广告黑名单的统计
* 对同一广告，一天内，点击超过100次的用户进行黑名单加入
*/
object Requirement1_BlackList {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setAppName("final_practice").setMaster("local[*]")
val ssc = new StreamingContext(sparkConf, Seconds(3))
//定义Kafka配置参数
val kafkaPara: Map[String, Object] = Map[String, Object](
//消费者服务器节点
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092,node2:9092,node3:9092",
//消费者组消费主题
ConsumerConfig.GROUP_ID_CONFIG -> "sparkstreamingpractice",
"key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)
//连接Kafka，获取消费数据
val kafkaDataDS = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferBrokers, //采集数据位置的策略，让框架自行匹配如何采集数据
ConsumerStrategies.Subscribe[String, String](Set("sparkstreamingpractice"), kafkaPara) //设置消费者相关参数
)
//对Kafka数据都封装到AdClickData模型中，让DS中每一条数据类型都是AdClickData实体
val adClickDataDS = kafkaDataDS.map(
kafkaData => {
val data = kafkaData.value() //获取数据value值
val splitData = data.split(" ")
//数据格式： timestamp area city userid adid（空格隔开）
AdClickData(splitData(0), splitData(2), splitData(3), splitData(4))
}
)
//TODO 周期性获取黑名单数据，校验用户是否位于黑名单：
// 若校验未通过，则过滤掉，该部分rdd不参与后续聚合
// 若校验通过，则该rdd会被保留在当前批次的DStream中，转换为(key, 1)格式，参与后续的聚合流程
//需要对DStream的每一条数据都进行校验
val filterDS = adClickDataDS.transform(
rdd => {
/*
通过jdbc周期性获取黑名单用户，对每个RDD进行校验
在Driver端统一执行，直接将不符合的数据拦截，防止其到Executor端再过滤
*/
//获取查询到的用户黑名单
val blackList = getBlackList
//对每个RDD执行过滤，过滤掉在黑名单的RDD（不在其中的才保留）
val filterRDD = rdd.filter(data => !blackList.contains(data.userid))
//对过滤后RDD进行统计转换,返回结果为以日期、user、广告为key的数据，计数值为初始值1
filterRDD.map(
data => {
//将时间戳格式进行转换
val format = new SimpleDateFormat("yyyy-MM-dd")
val day = format.format(new Date(data.timestamp.toLong))
((day, data.userid, data.ad), 1)
}
)
}
)
//TODO 校验通过（不位于黑名单）的用户，进行这一采集批次数据统计
val countDS = filterDS.reduceByKey(_ + _)
//TODO 校验这一批次聚合后数据每一条是否超过点击阈值
// 超过则拉入黑名单；不超过则更新数据库
// 同时，对于更新数据库的情况，若更新后数据超过阈值，同样需要拉入黑名单
countDS.foreachRDD( //对流中每一个RDD都执行校验 & 数据库更新
rdd => {
rdd.foreach(
{
case ((day, user, ad), count) => {
println(s"需要更新批次数据为：day:${day} ,user: ${user}, ad: ${ad}" )
//超过阈值，拉入黑名单
if (count >= 100) {
insertOrUpdateBlackList(user)
}
//未超过阈值，更新数据库
else {
insertOrUpdateUserAdCount(day, user, ad, count)
//更新后是否超过阈值？
val newCount = getCount(day, user, ad)
if (newCount >= 100) insertOrUpdateBlackList(user)
println("这一批次数据处理完成！")
println()
}
}
}
)
}
)
ssc.start()
ssc.awaitTermination()
}
/**
* 广告点击数据的数据模型实体，用于接收并封装Kafka数据
*/
private case class AdClickData(
timestamp: String, //时间戳
city: String, //城市信息
userid: String, //用户id
ad: String //广告信息
)
/**
* 从MySQL中获取黑名单用户
*
* @return
*/
private def getBlackList: ListBuffer[String] = {
//获取连接
val connection = JDBCUtil.getConnection
//准备查询SQL
val prep = connection.prepareStatement("select userid from black_list")
//执行查询SQL，获取结果
val resultSet = prep.executeQuery()
//准备SQL查询结果保存集合
val blackList = ListBuffer[String]()
//遍历查询结果，保存到准备的Set中
while (resultSet.next()) {
blackList.append(resultSet.getString("userid"))
}
//关闭各个连接资源
resultSet.close()
prep.close()
connection.close()
//返回查询结果保存集合
blackList
}
/**
* 将用户添加到黑名单
*
* @param user 本次要添加的用户数据
*/
private def insertOrUpdateBlackList(user: String): Unit = {
val connection = JDBCUtil.getConnection
//准备预编译SQL
//该SQL文中，若key重复（已经在表中存在）则执行update操作
val statement = connection.prepareStatement(
"""
|insert into black_list(userid) values (?)
|on duplicate key
|update userid = ?
|""".stripMargin)
//设置参数
statement.setString(1, user)
statement.setString(2, user)
//执行插入
statement.executeUpdate()
statement.close()
connection.close()
}
/**
* 将用户数据插入 / 更新到user_ad_count表中
*
* @param day 日期
* @param user 用户
* @param ad 点击广告
* @param count 点击次数
*/
private def insertOrUpdateUserAdCount(day: String, user: String, ad: String, count: Int): Unit = {
val connection = JDBCUtil.getConnection
//不存在，则将这条数据插入
//若存在，则更新数据
val statement = connection.prepareStatement(
"""
|insert into user_ad_count(dt, userid, adid, count) values(?, ?, ?, ?)
|on duplicate key
|update count = count + ?
|""".stripMargin)
statement.setString(1, day)
statement.setString(2, user)
statement.setString(3, ad)
statement.setInt(4, count)
statement.setInt(5, count)
statement.executeUpdate()
statement.close()
connection.close()
}
/**
* 从user_ad_count表中获取count字段值
* @param day 日期
* @param user 用户
* @param ad 点击广告
* @return
*/
def getCount(day: String, user: String, ad: String): Int = {
val connection = JDBCUtil.getConnection
val statement = connection.prepareStatement(
"""
|select
|count
|from user_ad_count
|where
|dt = ? and userid = ? and adid = ?
|""".stripMargin)
statement.setString(1, day)
statement.setString(2, user)
statement.setString(3, ad)
val resultSet = statement.executeQuery()
var count = 0
//若查询到了点击次数
if ( resultSet.next() )
count = resultSet.getInt("count")
resultSet.close()
statement.close()
connection.close()
count
}
}- 代码解析如下：
a 连接Kafka，消费本次需求topic的数据，同时将数据封装到模型实体中
//定义Kafka配置参数
val kafkaPara: Map[String, Object] = Map[String, Object](
//消费者服务器节点
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092,node2:9092,node3:9092",
//消费者组消费主题
ConsumerConfig.GROUP_ID_CONFIG -> "sparkstreamingpractice",
"key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)
//连接Kafka，获取消费数据
val kafkaDataDS = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferBrokers, //采集数据位置的策略，让框架自行匹配如何采集数据
ConsumerStrategies.Subscribe[String, String](Set("sparkstreamingpractice"), kafkaPara) //设置消费者相关参数
)
//对Kafka数据都封装到AdClickData模型中，让DS中每一条数据类型都是AdClickData实体
val adClickDataDS = kafkaDataDS.map(
kafkaData => {
val data = kafkaData.value() //获取数据value值
val splitData = data.split(" ")
//数据格式： timestamp area city userid adid（空格隔开）
AdClickData(splitData(0), splitData(2), splitData(3), splitData(4))
}
)我们为了后续数据处理的便利，将本次需求需要处理的Kafka数据进行了数据模型的封装：
/**
* 广告点击数据的数据模型实体，用于接收并封装Kafka数据
*/
private case class AdClickData(
timestamp: String, //时间戳
city: String, //城市信息
userid: String, //用户id
ad: String //广告信息
)因此，从Kafka消费后构成的DStream每个都是AdClickData的数据封装。
b 获取黑名单用户，对当前批次数据进行过滤，对过滤后的数据进行（key，1）的转换
//TODO 周期性获取黑名单数据，校验用户是否位于黑名单：
// 若校验未通过，则过滤掉，该部分rdd不参与后续聚合
// 若校验通过，则该rdd会被保留在当前批次的DStream中，转换为(key, 1)格式，参与后续的聚合流程
//需要对DStream的每一条数据都进行校验
val filterDS = adClickDataDS.transform(
rdd => {
/*
通过jdbc周期性获取黑名单用户，对每个RDD进行校验
在Driver端统一执行，直接将不符合的数据拦截，防止其到Executor端再过滤
*/
//获取查询到的用户黑名单
val blackList = getBlackList
//对每个RDD执行过滤，过滤掉在黑名单的RDD（不在其中的才保留）
val filterRDD = rdd.filter(data => !blackList.contains(data.userid))
//对过滤后RDD进行统计转换,返回结果为以日期、user、广告为key的数据，计数值为初始值1
filterRDD.map(
data => {
//将时间戳格式进行转换
val format = new SimpleDateFormat("yyyy-MM-dd")
val day = format.format(new Date(data.timestamp.toLong))
((day, data.userid, data.ad), 1)
}
)
}
)由于我们需要对每一批次的DStream都进行用户是否位于黑名单的校验，因此这是一个周期性操作，即这个黑名单用户的生成需要每一批DStream都进行一次；同时我们不想要这个黑名单在Executor端生成，因此我们需要在Driver端周期性执行黑名单用户生成的操作。
基于这些需求，我们通过DStream.transform操作，以在对每一批DStream操作的时候，都在Driver端生成黑名单，然后再在过滤时的Executor使用Driver端的黑名单。
对MySQL黑名单数据获取的逻辑封装如下：
/**
* 从MySQL中获取黑名单用户
* @return
*/
private def getBlackList: ListBuffer[String] = {
//获取连接
val connection = JDBCUtil.getConnection
//准备查询SQL
val prep = connection.prepareStatement("select userid from black_list")
//执行查询SQL，获取结果
val resultSet = prep.executeQuery()
//准备SQL查询结果保存集合
var blackList = ListBuffer[String]()
//遍历查询结果，保存到准备的Set中
while ( resultSet.next() ) {
blackList.append(resultSet.getString("userid"))
}
//关闭各个连接资源
resultSet.close()
prep.close()
connection.close()
//返回查询结果保存集合
blackList
}c 对上一步过滤后的DStream进行wordcount聚合
//TODO 校验通过（不位于黑名单）的用户，进行这一采集批次数据统计
val countDS = filterDS.reduceByKey(_ + _)实际上，在transform内部也能够实现对本批次RDD的reduceByKey。但是为职责分离 & 代码解耦，我们令transform只负责数据过滤 & 转换，聚合操作由新DStream处理。
d 对聚合后的批数据进行校验，判断是否仍然需要插入黑名单
//TODO 校验这一批次聚合后数据每一条是否超过点击阈值
// 超过则拉入黑名单；不超过则更新数据库
// 同时，对于更新数据库的情况，若更新后数据超过阈值，同样需要拉入黑名单
countDS.foreachRDD( //对流中每一个RDD都执行校验 & 数据库更新
rdd => {
rdd.foreach(
{
case ((day, user, ad), count) => {
println(s"需要更新批次数据为：day:${day} ,user: ${user}, ad: ${ad}" )
//超过阈值，拉入黑名单
if (count >= 100) {
insertIntoBlackList(user)
}
//未超过阈值，更新数据库
else {
insertOrUpdateUserAdCount(day, user, ad, count)
//更新后是否超过阈值？
val newCount = getCount(day, user, ad)
if (newCount >= 100) insertIntoBlackList(user)
println("这一批次数据处理完成！")
println()
}
}
}
)
}
)这一步骤需要进行两次校验：若rdd内的聚合数据count结果已经超过阈值，则直接拉入黑名单；若rdd内聚合数据未超过阈值，则插入数据库的user_ad_count表，然后判断插入后结果是否超过阈值，超过则同样需要拉入黑名单。
该步骤用到的数据库操作函数如下：
① insertIntoBlackList——将用户 插入 / 更新 到黑名单table
/**
* 将用户添加到黑名单
*
* @param user 本次要添加的用户数据
*/
private def insertOrUpdateBlackList(user: String): Unit = {
val connection = JDBCUtil.getConnection
//准备预编译SQL
//该SQL文中，若key重复（已经在表中存在）则执行update操作
val statement = connection.prepareStatement(
"""
|insert into black_list(userid) values (?)
|on duplicate key
|update userid = ?
|""".stripMargin)
//设置参数
statement.setString(1, user)
statement.setString(2, user)
//执行插入
statement.executeUpdate()
statement.close()
connection.close()
}因为需要根据我们传入的参数执行插入操作，所以我们通过prepareStatement来准备预编译SQL，来插入我们的参数。
同时，由于黑名单用户可能已经存在于table中，我们不应该对同一userid反复执行insert操作，所以我们通过SQL文中的”on duplicate key update”语句，这条语句能够允许我们在存在相同key的数据时，对指定字段执行update操作。
② insertOrUpdateUserAdCount——将聚合结果 插入 / 更新 到用户点击table
/**
* 将用户数据插入 / 更新到user_ad_count表中
*
* @param day 日期
* @param user 用户
* @param ad 点击广告
* @param count 点击次数
*/
private def insertOrUpdateUserAdCount(day: String, user: String, ad: String, count: Int): Unit = {
val connection = JDBCUtil.getConnection
//不存在，则将这条数据插入
//若存在，则更新数据
val statement = connection.prepareStatement(
"""
|insert into user_ad_count(dt, userid, adid, count) values(?, ?, ?, ?)
|on duplicate key
|update count = count + ?
|""".stripMargin)
statement.setString(1, day)
statement.setString(2, user)
statement.setString(3, ad)
statement.setInt(4, count)
statement.setInt(5, count)
statement.executeUpdate()
statement.close()
connection.close()
}同样的，由于点击数据可能已经存在，这种情况下我们需要对count进行update而非insert，因此通过”on duplicate key update”来进行。
③ getCount——从数据库中获取当前条件下用户的点击次数结果
/**
* 从user_ad_count表中获取count字段值
* @param day 日期
* @param user 用户
* @param ad 点击广告
* @return
*/
def getCount(day: String, user: String, ad: String): Int = {
val connection = JDBCUtil.getConnection
val statement = connection.prepareStatement(
"""
|select
|count
|from user_ad_count
|where
|dt = ? and userid = ? and adid = ?
|""".stripMargin)
statement.setString(1, day)
statement.setString(2, user)
statement.setString(3, ad)
val resultSet = statement.executeQuery()
var count = 0
//若查询到了点击次数
if ( resultSet.next() )
count = resultSet.getInt("count")
resultSet.close()
statement.close()
connection.close()
count
}（4）代码测试
- 运行模拟数据生成模块 & 需求一模块，查看控制台输出：
- 查看数据库两张table的数据情况：
需要说明的是，由于我们设置模拟数据中的userid 只有6个，adid同样只有6个，所以仅会生成 6 * 6种不同的广告点击数据，因此user_ad_count表最多不会超过36条数据。
（5）代码优化
a 优化一：将JDBC操作进行封装
不难发现，在我们实现的代码中，涉及到很多对JDBC的操作（比如insertOrUpdateBlackList等），在每一个操作中我们都需要执行 建立connection、准备statement、执行statement、关闭连接 一系列操作，难免有些繁琐，且代码重复严重。
我们可以将这部分内容同一封装到JDBCUtil工具类，在不同方法使用时，只需要传入 连接参数、sql文、预编译SQL参数 即可。
- 对插入操作在JDBCUtil的封装：
/**
* 同一执行更新操作，对单条数据进行插入
* @param connectino MySQL连接对象
* @param sql 执行的sql文
* @param params 预编译SQL的参数列表
* @return
*/
def executeUpdate(connectino: Connection, sql: String, params: Array[Any]): Int = {
var updateCount = 0 //记录更新数量
var prepareStatement: PreparedStatement = null
try {
connectino.setAutoCommit(false) //防止自动提交事务
prepareStatement = connectino.prepareStatement(sql) //将sql文参数赋值给预编译SQL
//为预编译SQL赋予参数
if (params != null && params.length > 0) {
for (i <- params.indices) {
prepareStatement.setObject(i + 1, params(i)) //预编译SQL参数从1开始，参数数组从0索引开始
}
}
updateCount = prepareStatement.executeUpdate()
connectino.commit() //执行成功，提交事务
prepareStatement.close()
} catch {
case e: Exception => e.printStackTrace()
}
//将更新数据影响数量返回
updateCount
}我们在JDBCUtil中新增方法，该方法用于将数据插入到MySQL中。我们只需要为该方法传入三个参数：connection连接对象、需要执行的updateSQL字符串、预编译SQL需要的参数列表，就可以在涉及到插入的地方直接调用该方法进行插入。
需要注意，prepareStatement的参数从1开始计算（占位符），而params参数列表则从0索引开始计算，二者需要对应。
- 这是使用统一插入操作的方法示例：
/**
* 将用户添加到黑名单
*
* @param user 本次要添加的用户数据
*/
private def insertOrUpdateBlackList(user: String): Unit = {
//准备更新参数
val connection = JDBCUtil.getConnection
val sql = """
|insert into black_list(userid) values (?)
|on duplicate key
|update userid = ?
|""".stripMargin
val params: Array[Any] = Array( user, user )
//执行更新操作
JDBCUtil.executeUpdate(connection, sql, params)
//关闭连接
connection.close()
}
/**
* 将用户数据插入 / 更新到user_ad_count表中
* @param day 日期
* @param user 用户
* @param ad 点击广告
* @param count 点击次数
*/
private def insertOrUpdateUserAdCount(day: String, user: String, ad: String, count: Int): Unit = {
val connection = JDBCUtil.getConnection
//不存在，则将这条数据插入
//若存在，则更新数据
val sql = """
|insert into user_ad_count(dt, userid, adid, count) values(?, ?, ?, ?)
|on duplicate key
|update count = count + ?
|""".stripMargin
val params: Array[Any] = Array(day, user, ad, count, count)
JDBCUtil.executeUpdate(connection, sql, params)
connection.close()
}由于这两个插入的SQL均包含对于数据是否存在的判断，因此此处主要封装插入语句（代码中大部分也都是插入语句）。
b 优化二：优化connection连接对象的创建
① 让方法使用传入的connection参数，避免每个方法都要创建 & 关闭
在前面代码中，我们在每一个方法中都进行了connection的创建，这显然是不合理的。我们首先需要让这个代码块中所有方法都共有同一个connection。否则，每个方法都要创建 & 关闭connection，这会影响性能。
- 首先，为每个方法添加connection参数，使用我们调用时传入的connection对象，不再自行创建：
/**
* 从MySQL中获取黑名单用户
*
* @return
*/
private def getBlackList(connection: Connection): ListBuffer[String] = {
//准备查询SQL
val prep = connection.prepareStatement("select userid from black_list")
//执行查询SQL，获取结果
val resultSet = prep.executeQuery()
//准备SQL查询结果保存集合
val blackList = ListBuffer[String]()
//遍历查询结果，保存到准备的Set中
while (resultSet.next()) {
blackList.append(resultSet.getString("userid"))
}
//关闭各个连接资源
resultSet.close()
prep.close()
//返回查询结果保存集合
blackList
}
/**
* 将用户添加到黑名单
*
* @param user 本次要添加的用户数据
*/
private def insertOrUpdateBlackList(connection: Connection, user: String): Unit = {
val sql =
"""
|insert into black_list(userid) values (?)
|on duplicate key
|update userid = ?
|""".stripMargin
val params: Array[Any] = Array(user, user)
//执行更新操作
JDBCUtil.executeUpdate(connection, sql, params)
}
/**
* 将用户数据插入 / 更新到user_ad_count表中
*
* @param day 日期
* @param user 用户
* @param ad 点击广告
* @param count 点击次数
*/
private def insertOrUpdateUserAdCount(connection: Connection, day: String, user: String, ad: String, count: Int): Unit = {
//不存在，则将这条数据插入
//若存在，则更新数据
val sql =
"""
|insert into user_ad_count(dt, userid, adid, count) values(?, ?, ?, ?)
|on duplicate key
|update count = count + ?
|""".stripMargin
val params: Array[Any] = Array(day, user, ad, count, count)
JDBCUtil.executeUpdate(connection, sql, params)
}
/**
* 从user_ad_count表中获取count字段值
*
* @param day 日期
* @param user 用户
* @param ad 点击广告
* @return
*/
def getCount(connection: Connection, day: String, user: String, ad: String): Int = {
val statement = connection.prepareStatement(
"""
|select
|count
|from user_ad_count
|where
|dt = ? and userid = ? and adid = ?
|""".stripMargin)
statement.setString(1, day)
statement.setString(2, user)
statement.setString(3, ad)
val resultSet = statement.executeQuery()
var count = 0
//若查询到了点击次数
if (resultSet.next())
count = resultSet.getInt("count")
resultSet.close()
statement.close()
count
}- 然后我们就可以在对每个RDD操作时，创建这个connection，让每个方法均使用这个connection，且由我们统一进行close：
//TODO 校验这一批次聚合后数据每一条是否超过点击阈值
// 超过则拉入黑名单；不超过则更新数据库
// 同时，对于更新数据库的情况，若更新后数据超过阈值，同样需要拉入黑名单
countDS.foreachRDD( //对流中每一个RDD都执行校验 & 数据库更新
rdd => {
rdd.foreach(
{
case ((day, user, ad), count) => {
val connection = JDBCUtil.getConnection
println(s"需要更新批次数据为：day:${day} ,user: ${user}, ad: ${ad}" )
//超过阈值，拉入黑名单
if (count >= 100) {
insertOrUpdateBlackList(connection, user)
}
//未超过阈值，更新数据库
else {
insertOrUpdateUserAdCount(connection, day, user, ad, count)
//更新后是否超过阈值？
val newCount = getCount(connection, day, user, ad)
if (newCount >= 100) insertOrUpdateBlackList(connection, user)
println("这一批次数据处理完成！")
println()
}
connection.close()
}
}
)
}
)② 优化connection的创建，避免对RDD每一条数据都要创建一次connection
在避免每个方法都要自行创建 & 关闭connection后，还剩下一个问题，那就是目前connection的创建位置，如下图：
我们知道，rdd.foreach实际上是获取rdd中每一条数据。当前connection的创建实际上会被每一条数据所执行。也就是说我们之前的优化避免了每一条数据调用每一次方法时都需要创建connection，现在我们需要优化，来避免对每一条数据都创建一次connection。我们的目标是：至少让RDD中的一批数据都可以共用同一个connection。
我们可能首先会想到，既然connection是在模式匹配内创建，这导致对每一条数据都进行一次创建，那么我们能否将connection创建位置调整到rdd.foreach之前，让这个rdd中所有数据都能够使用这个connection？比如放到这里：
那么答案是：不行，且绝对不行。最核心的原因是，rdd内的共享数据需要通过闭包传递，闭包传递的参数需要能够被序列化，而connection无法被序列化。
那么应该如何解决这个问题呢？我们就是想要让至少一批数据能够共有同一个connection，不要一条数据创建一个。
回想我们之前在SparkCore中所学到的partitions概念，我们可以取出RDD中一个分区的所有数据，直接对这批数据进行操作。那么，我们为什么不使用这个思路，取出一个partitons的数据，让这批数据共享一个connection呢？这虽然无法做到让rdd中每一个数据都使用同一个connection，但足以让rdd中 1/ 分区数 的数据使用同一个connection，这照样能够提升性能！
我们可以不再使用RDD.foreach方法，因为这会取出RDD中每一条数据而非每一个分区的数据。我们应该使用RDD.foreachPartitions方法，来实现我们的需求：
//一次性对RDD内一个分区的一批数据进行操作
rdd.foreachPartition(
iter => {
val connection = JDBCUtil.getConnection
//对这一分区中每一条数据都进行相关业务逻辑
iter.foreach(
{
case ((day, user, ad), count) => {
println(s"需要更新批次数据为：day:${day} ,user: ${user}, ad: ${ad}")
//超过阈值，拉入黑名单
if (count >= 100) {
insertOrUpdateBlackList(connection, user)
}
//未超过阈值，更新数据库
else {
insertOrUpdateUserAdCount(connection, day, user, ad, count)
//更新后是否超过阈值？
val newCount = getCount(connection, day, user, ad)
if (newCount >= 100) insertOrUpdateBlackList(connection, user)
println("这一批次数据处理完成！")
println()
}
}
}
)
connection.close()
}
)使用foreachPartitions会比较复杂，因为需要取得这个分区内所有数据的Iterator，然后再遍历这个Iterator，对每一条数据进行处理。
那么，为什么 不能在rdd.foreach之前设置统一connection，但是在 iter.foreach之前却可以呢？因为，迭代器在内存中，对于这一批迭代器数据的处理不涉及闭包，不用进行序列化，connection就可以被这一批数据共享。
（6）完整代码
a JDBCUtil
package main.spark_streaming.final_practice.util
import com.alibaba.druid.pool.DruidDataSourceFactory
import java.sql.{Connection, PreparedStatement}
import java.util.Properties
import javax.sql.DataSource
/**
* JDBC相关操作的封装工具类
*/
object JDBCUtil {
var dataSource: DataSource = init()
/**
* 初始化数据库连接池
* @return
*/
private def init(): DataSource = {
val url = "jdbc:mysql://localhost:3306/spark_streaming_practice?useUnicode=true&characterEncoding=UTF-8&useSSL=false&user=root&password=D200504193010"
val driver = "com.mysql.cj.jdbc.Driver"
val properties = new Properties()
//设置JDBC的连接参数
properties.setProperty("url", url)
properties.setProperty("driver", driver)
properties.setProperty("maxActive", "50") //最大连接池数量
properties.setProperty("connectionProperties", "useUnicode=true;characterEncoding=UTF-8")
DruidDataSourceFactory.createDataSource(properties)
}
/**
* 获取MySQL连接
* @return
*/
def getConnection: Connection = {
dataSource.getConnection
}
/**
* 同一执行更新操作，对单条数据进行插入
* @param connectino MySQL连接对象
* @param sql 执行的sql文
* @param params 预编译SQL的参数列表
* @return 更新的数据数量
*/
def executeUpdate(connectino: Connection, sql: String, params: Array[Any]): Int = {
var updateCount = 0 //记录更新数量
var prepareStatement: PreparedStatement = null
try {
connectino.setAutoCommit(false) //防止自动提交事务
prepareStatement = connectino.prepareStatement(sql) //将sql文参数赋值给预编译SQL
//为预编译SQL赋予参数
if (params != null && params.length > 0) {
for (i <- params.indices) {
prepareStatement.setObject(i + 1, params(i)) //预编译SQL参数从1开始，参数数组从0索引开始
}
}
updateCount = prepareStatement.executeUpdate()
connectino.commit() //执行成功，提交事务
prepareStatement.close()
} catch {
case e: Exception => e.printStackTrace()
}
//将更新数据影响数量返回
updateCount
}
/**
* 判断数据是否存在
* @param connection MySQL连接对象
* @param sql 要执行的查询SQL文
* @param params 预编译SQL的参数
* @return
*/
def isExist(connection: Connection, sql: String, params: Array[Any]): Boolean = {
var isExist: Boolean = false //是否存在数据的记录
var prepareStatement: PreparedStatement = null
try {
prepareStatement = connection.prepareStatement(sql)
if (params != null && params.length > 0) {
for (i <- params.indices)
prepareStatement.setObject(i + 1, params(i))
}
//若查询的resultSet有内容，则为true
isExist = prepareStatement.executeQuery().next()
prepareStatement.close()
} catch {
case e: Exception => e.printStackTrace()
}
isExist
}
}b MockData
package main.spark_streaming.final_practice
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import java.util.Properties
import scala.collection.mutable.ListBuffer
import scala.util.Random
/**
* 案例流程：//Application => Kafka => SparkStreaming => 统计分析
* 该模块为Application，用于生成模拟数据，推送到Kafka，让SparkStreaming实时消费
*/
object MockData {
def main(args: Array[String]): Unit = {
//生成模拟数据
//数据格式： timestamp area city userid adid（空格隔开）
//数据含义： 时间戳 区域 城市 用户 广告
//创建一个Kafka生产者
val producer = createKafkaProducer("node1:9092")
//生产数据到Kafka
produceToKafka(producer)
}
/**
* 生成模拟数据，一批生成30条数据
* 生成的数据每一条都遵循格式：timestamp area city userid adid（空格隔开）
* @return
*/
def generatingMockData(): ListBuffer[String] = {
val mockDataList = ListBuffer[String]() //用于存储生成数据的集合
val areaList = List("华东", "华北", "华南", "华中") //area列表，用于随机数据的area参数
val cityList = List("北京", "上海", "深圳", "广州", "厦门") //city列表，用于随机数据的city参数
for (i <- 1 to 30) {
val area = areaList(new Random().nextInt(4)) //随机选一个area
val city = cityList(new Random().nextInt(5)) //随机选一个city
val userid = new Random().nextInt(6) //随机生成一个user
val adid = new Random().nextInt(6) //随机生成一个点击广告
mockDataList.append(
s"${System.currentTimeMillis()} ${area} ${city} ${userid} ${adid}"
)
}
mockDataList
}
/**
* 创建Kafka生成者
* @param broker Kafka leader服务器节点url
* @return
*/
def createKafkaProducer(broker: String): KafkaProducer[String, String] = {
// 创建配置对象
val prop = new Properties()
// 添加配置
prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker)
prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer")
prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer")
// 根据配置创建Kafka生产者
new KafkaProducer[String, String](prop)
}
/**
* 向Kafka推送数据的核心流程
* @param kafkaProducer Kafka生产者，用于推送数据
*/
def produceToKafka(kafkaProducer: KafkaProducer[String, String]): Unit = {
while (true) {
//调用生产随机数据，并将所有内容推送到Kafka中
generatingMockData().foreach(
data => {
println("生成的模拟数据为：" + data)
//创建一个生产者record，指定生产到的 topic & 数据内容，用于封装向Kafka发送的数据
val record = new ProducerRecord[String, String]("sparkstreamingpractice", data)
//向Kafka推送数据
kafkaProducer.send(record)
}
)
//每2s生产、发送一次即可
Thread.sleep(2000)
}
}
}c Requirement1_BlackList
package main.spark_streaming.final_practice
import main.spark_streaming.final_practice.util.JDBCUtil
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import java.sql.Connection
import java.text.SimpleDateFormat
import java.util.Date
import scala.collection.mutable.ListBuffer
/**
* 广告黑名单的统计
* 对同一广告，一天内，点击超过100次的用户进行黑名单加入
*/
object Requirement1_BlackList {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setAppName("final_practice").setMaster("local[*]")
val ssc = new StreamingContext(sparkConf, Seconds(3))
//定义Kafka配置参数
val kafkaPara: Map[String, Object] = Map[String, Object](
//消费者服务器节点
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092,node2:9092,node3:9092",
//消费者组消费主题
ConsumerConfig.GROUP_ID_CONFIG -> "sparkstreamingpractice",
"key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)
//连接Kafka，获取消费数据
val kafkaDataDS = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferBrokers, //采集数据位置的策略，让框架自行匹配如何采集数据
ConsumerStrategies.Subscribe[String, String](Set("sparkstreamingpractice"), kafkaPara) //设置消费者相关参数
)
//对Kafka数据都封装到AdClickData模型中，让DS中每一条数据类型都是AdClickData实体
val adClickDataDS = kafkaDataDS.map(
kafkaData => {
val data = kafkaData.value() //获取数据value值
val splitData = data.split(" ")
//数据格式： timestamp area city userid adid（空格隔开）
AdClickData(splitData(0), splitData(2), splitData(3), splitData(4))
}
)
//TODO 周期性获取黑名单数据，校验用户是否位于黑名单：
// 若校验未通过，则过滤掉，该部分rdd不参与后续聚合
// 若校验通过，则该rdd会被保留在当前批次的DStream中，转换为(key, 1)格式，参与后续的聚合流程
//需要对DStream的每一条数据都进行校验
val filterDS = adClickDataDS.transform(
rdd => {
/*
通过jdbc周期性获取黑名单用户，对每个RDD进行校验
在Driver端统一执行，直接将不符合的数据拦截，防止其到Executor端再过滤
*/
//获取查询到的用户黑名单
val connection = JDBCUtil.getConnection
val blackList = getBlackList(connection)
connection.close()
//对每个RDD执行过滤，过滤掉在黑名单的RDD（不在其中的才保留）
val filterRDD = rdd.filter(data => !blackList.contains(data.userid))
//对过滤后RDD进行统计转换,返回结果为以日期、user、广告为key的数据，计数值为初始值1
filterRDD.map(
data => {
//将时间戳格式进行转换
val format = new SimpleDateFormat("yyyy-MM-dd")
val day = format.format(new Date(data.timestamp.toLong))
((day, data.userid, data.ad), 1)
}
)
}
)
//TODO 校验通过（不位于黑名单）的用户，进行这一采集批次数据统计
val countDS = filterDS.reduceByKey(_ + _)
//TODO 校验这一批次聚合后数据每一条是否超过点击阈值
// 超过则拉入黑名单；不超过则更新数据库
// 同时，对于更新数据库的情况，若更新后数据超过阈值，同样需要拉入黑名单
countDS.foreachRDD( //对流中每一个RDD都执行校验 & 数据库更新
rdd => {
//val connection = JDBCUtil.getConnection
//一次性对RDD内一个分区的一批数据进行操作
rdd.foreachPartition(
iter => {
val connection = JDBCUtil.getConnection
//对这一分区中每一条数据都进行相关业务逻辑
iter.foreach(
{
case ((day, user, ad), count) => {
println(s"需要更新批次数据为：day:${day} ,user: ${user}, ad: ${ad}")
//超过阈值，拉入黑名单
if (count >= 100) {
insertOrUpdateBlackList(connection, user)
}
//未超过阈值，更新数据库
else {
insertOrUpdateUserAdCount(connection, day, user, ad, count)
//更新后是否超过阈值？
val newCount = getCount(connection, day, user, ad)
if (newCount >= 100) insertOrUpdateBlackList(connection, user)
println("这一批次数据处理完成！")
println()
}
}
}
)
connection.close()
}
)
}
)
ssc.start()
ssc.awaitTermination()
}
/**
* 广告点击数据的数据模型实体，用于接收并封装Kafka数据
*/
private case class AdClickData(
timestamp: String, //时间戳
city: String, //城市信息
userid: String, //用户id
ad: String //广告信息
)
/**
* 从MySQL中获取黑名单用户
*
* @return
*/
private def getBlackList(connection: Connection): ListBuffer[String] = {
//准备查询SQL
val prep = connection.prepareStatement("select userid from black_list")
//执行查询SQL，获取结果
val resultSet = prep.executeQuery()
//准备SQL查询结果保存集合
val blackList = ListBuffer[String]()
//遍历查询结果，保存到准备的Set中
while (resultSet.next()) {
blackList.append(resultSet.getString("userid"))
}
//关闭各个连接资源
resultSet.close()
prep.close()
//返回查询结果保存集合
blackList
}
/**
* 将用户添加到黑名单
*
* @param user 本次要添加的用户数据
*/
private def insertOrUpdateBlackList(connection: Connection, user: String): Unit = {
val sql =
"""
|insert into black_list(userid) values (?)
|on duplicate key
|update userid = ?
|""".stripMargin
val params: Array[Any] = Array(user, user)
//执行更新操作
JDBCUtil.executeUpdate(connection, sql, params)
}
/**
* 将用户数据插入 / 更新到user_ad_count表中
*
* @param day 日期
* @param user 用户
* @param ad 点击广告
* @param count 点击次数
*/
private def insertOrUpdateUserAdCount(connection: Connection, day: String, user: String, ad: String, count: Int): Unit = {
//不存在，则将这条数据插入
//若存在，则更新数据
val sql =
"""
|insert into user_ad_count(dt, userid, adid, count) values(?, ?, ?, ?)
|on duplicate key
|update count = count + ?
|""".stripMargin
val params: Array[Any] = Array(day, user, ad, count, count)
JDBCUtil.executeUpdate(connection, sql, params)
}
/**
* 从user_ad_count表中获取count字段值
*
* @param day 日期
* @param user 用户
* @param ad 点击广告
* @return
*/
def getCount(connection: Connection, day: String, user: String, ad: String): Int = {
val statement = connection.prepareStatement(
"""
|select
|count
|from user_ad_count
|where
|dt = ? and userid = ? and adid = ?
|""".stripMargin)
statement.setString(1, day)
statement.setString(2, user)
statement.setString(3, ad)
val resultSet = statement.executeQuery()
var count = 0
//若查询到了点击次数
if (resultSet.next())
count = resultSet.getInt("count")
resultSet.close()
statement.close()
count
}
}7.9.3 需求二
（1）思路分析
（2）准备工作——MySQL建表
user spark_streaming_practice;
create table if not exists area_city_ad_count
(
dt varchar(255) not null,
area varchar(255) not null,
city varchar(255) not null,
adid varchar(1) not null,
count bigint null,
primary key (dt, area, city, adid)
);（3）代码实现
- 完整业务代码如下：
package main.spark_streaming.final_practice.code
import main.spark_streaming.final_practice.util.JDBCUtil
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import java.sql.Connection
import java.text.SimpleDateFormat
import java.util.Date
object Requirement2_AdCount {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setAppName("final_practice").setMaster("local[*]")
val ssc = new StreamingContext(sparkConf, Seconds(3))
//创建Kafka消费者
//定义Kafka配置参数
val kafkaPara: Map[String, Object] = Map[String, Object](
//消费者服务器节点
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092,node2:9092,node3:9092",
//消费者组消费主题
ConsumerConfig.GROUP_ID_CONFIG -> "sparkstreamingpractice",
"key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)
//连接Kafka，获取消费数据
val kafkaDataDS = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferBrokers, //采集数据位置的策略，让框架自行匹配如何采集数据
ConsumerStrategies.Subscribe[String, String](Set("sparkstreamingpractice"), kafkaPara) //设置消费者相关参数
)
//将Kafka流数据封装到数据模型中
val adClickDataDS = kafkaDataDS.map(
kafkaData => {
val value = kafkaData.value() //获取数据的值
val splitData = value.split(" ")
ClickLogModule(splitData(0), splitData(1), splitData(2), splitData(4))
}
)
//对数据转换为(key, 1)的形式，即：((timestamp, city, area, ad), 1)
val mapDS = adClickDataDS.map(
data => {
val timestamp = data.timeStamp
//将时间戳格式进行转换
val format = new SimpleDateFormat("yyyy-MM-dd")
val day = format.format(new Date(timestamp.toLong))
((day, data.area, data.city, data.ad), 1)
}
)
//进行按key的聚合操作
val reduceDS = mapDS.reduceByKey(_ + _)
//将数据写入数据库
//把这一批每个RDD数据都写入MySQL
reduceDS.foreachRDD(
rdd => {
//同样的，为了让尽可能多的数据共享同一个connection，对rdd分区数据进行统一操作
rdd.foreachPartition(
iter => {
val connection = JDBCUtil.getConnection
iter.foreach(
{
//匹配数据同时对元组进行解耦
case ((day, area, city, ad), count) => {
println(s"开始处理聚合数据！本次处理数据为 day: ${day}, area: ${area}, city: ${city}, ad: ${ad}, count: ${count}")
//插入数据库
val result = insertOrUpdateAreaCityAdCount(connection, day, area, city, ad, count)
println(s"此次更新了${result}条数据")
}
}
)
println("这批次数据处理完成，关闭连接！")
connection.close()
}
)
}
)
ssc.start()
ssc.awaitTermination()
}
/**
* 用于封装Kafka生成数据的数据模型
* @param timeStamp 时间戳
* @param area 地区
* @param city 城市
* @param ad 点击广告
*/
case class ClickLogModule(
timeStamp: String,
area: String,
city: String,
ad: String
)
/**
* 将数据 插入 / 更新 到数据库中
* @param connection 数据库连接对象
* @param day 日期参数
* @param area 地区参数
* @param city 城市参数
* @param ad 点击广告
* @param count 统计
* @return
*/
private def insertOrUpdateAreaCityAdCount(
connection: Connection,
day: String,
area: String,
city: String,
ad: String,
count: Int
): Int = {
//准备参数列表
val params = Array(day, area, city, ad, count, count)
//需要准备的插入SQL文
val sql =
"""
|insert into
|area_city_ad_count(dt, area, city, adid, count)
|values(?, ?, ?, ? ,?)
|on duplicate key
|update count = count + ?
|""".stripMargin
//获取修改的条数
val result = JDBCUtil.executeUpdate(connection, sql, params)
result
}
}a 消费Kafka数据，将其读取到SparkStreaming中，并将每条数据都封装到对应数据模型中
//创建Kafka消费者
//定义Kafka配置参数
val kafkaPara: Map[String, Object] = Map[String, Object](
//消费者服务器节点
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092,node2:9092,node3:9092",
//消费者组消费主题
ConsumerConfig.GROUP_ID_CONFIG -> "sparkstreamingpractice",
"key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)
//连接Kafka，获取消费数据
val kafkaDataDS = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferBrokers, //采集数据位置的策略，让框架自行匹配如何采集数据
ConsumerStrategies.Subscribe[String, String](Set("sparkstreamingpractice"), kafkaPara) //设置消费者相关参数
)
//将Kafka流数据封装到数据模型中
val adClickDataDS = kafkaDataDS.map(
kafkaData => {
val value = kafkaData.value() //获取数据的值
val splitData = value.split(" ")
ClickLogModule(splitData(0), splitData(1), splitData(2), splitData(4))
}
)准备的数据模型实体如下：
/**
* 用于封装Kafka生成数据的数据模型
* @param timeStamp 时间戳
* @param area 地区
* @param city 城市
* @param ad 点击广告
*/
case class ClickLogModule(
timeStamp: String,
area: String,
city: String,
ad: String
)b 对数据进行map转换，转换为(key, 1)的形式，便于后续聚合统计
//对数据转换为(key, 1)的形式，即：((timestamp, city, area, ad), 1)
val mapDS = adClickDataDS.map(
data => {
val timestamp = data.timeStamp
//将时间戳格式进行转换
val format = new SimpleDateFormat("yyyy-MM-dd")
val day = format.format(new Date(timestamp.toLong))
((day, data.area, data.city, data.ad), 1)
}
)c 将转换后数据按key进行聚合统计
//进行按key的聚合操作
val reduceDS = mapDS.reduceByKey(_ + _)d 将统计后的数据插入到数据库
//将数据写入数据库
//把这一批每个RDD数据都写入MySQL
reduceDS.foreachRDD(
rdd => {
//同样的，为了让尽可能多的数据共享同一个connection，对rdd分区数据进行统一操作
rdd.foreachPartition(
iter => {
val connection = JDBCUtil.getConnection
iter.foreach(
{
//匹配数据同时对元组进行解耦
case ((day, area, city, ad), count) => {
println(s"开始处理聚合数据！本次处理数据为 day: ${day}, area: ${area}, city: ${city}, ad: ${ad}, count: ${count}")
//插入数据库
val result = insertOrUpdateAreaCityAdCount(connection, day, area, city, ad, count)
println(s"此次更新了${result}条数据")
}
}
)
println("这批次数据处理完成，关闭连接！")
connection.close()
}
)
}
)同样的，为了增强程序性能，在插入数据时采用rdd.foreachPartitions，让同一分区的所有数据共享同一个connection。
用于执行插入操作的方法如下：
/**
* 将数据 插入 / 更新 到数据库中
* @param connection 数据库连接对象
* @param day 日期参数
* @param area 地区参数
* @param city 城市参数
* @param ad 点击广告
* @param count 统计
* @return
*/
private def insertOrUpdateAreaCityAdCount(
connection: Connection,
day: String,
area: String,
city: String,
ad: String,
count: Int
): Int = {
//准备参数列表
val params = Array(day, area, city, ad, count, count)
//需要准备的插入SQL文
val sql =
"""
|insert into
|area_city_ad_count(dt, area, city, adid, count)
|values(?, ?, ?, ? ,?)
|on duplicate key
|update count = count + ?
|""".stripMargin
//获取修改的条数
val result = JDBCUtil.executeUpdate(connection, sql, params)
result
}整体仍然沿用了JDBCUtil封装的executeUpdate方法。该方法中仅提供sql文 & 预编译SQL参数，具体操作（statement等对象创建 & 查询）在executeUpdate中实现。executeUpdate操作详情参考7.9.2-（5）-a。
（4）代码测试
运行业务代码 & 数据生成模块：
查看数据库：
7.9.4 需求三
我们统计以10分钟为区间的广告点击情况，如下：
同时，我们想要每次统计1分钟的数据，且每10s统计一次。
（1）思路分析
首先，由于需要统计的是以10分钟为区间的数据，我们就需要先将数据都转换为如下格式：12:53, 12:56, 12:59 -> 12:50，11:11, 11:17, 11:12 -> 11:10…其实很简单，取出分钟的十分位构成新的时间即可。
然后就是我们统计频率的需求：每次统计1分钟的数据，每10s统计一次。这显然是对窗口大小 & 窗口步长提出了要求，我们聚合时通过window进行即可。
整体来看不算特别难的需求，重点是日期转换 & 滑动窗口操作。
（2）代码实现
- 完整业务代码如下：
package main.spark_streaming.final_practice.code
import main.spark_streaming.final_practice.code.Requirement2_AdCount.ClickLogModule
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.{SparkConf, streaming}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import java.text.SimpleDateFormat
import java.util.Date
object Requirement3_Latest10MintuesAdClick {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setAppName("final_practice").setMaster("local[*]")
val ssc = new StreamingContext(sparkConf, Seconds(5))
//创建Kafka消费者
//定义Kafka配置参数
val kafkaPara: Map[String, Object] = Map[String, Object](
//消费者服务器节点
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092,node2:9092,node3:9092",
//消费者组消费主题
ConsumerConfig.GROUP_ID_CONFIG -> "sparkstreamingpractice",
"key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)
//连接Kafka，获取消费数据
val kafkaDataDS = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferBrokers, //采集数据位置的策略，让框架自行匹配如何采集数据
ConsumerStrategies.Subscribe[String, String](Set("sparkstreamingpractice"), kafkaPara) //设置消费者相关参数
)
//将Kafka流数据封装到数据模型中
val adClickDataDS = kafkaDataDS.map(
kafkaData => {
val value = kafkaData.value() //获取数据的值
val splitData = value.split(" ")
//数据格式： timestamp area city userid adid（空格隔开）
AdClickData(splitData(0), splitData(4))
}
)
//TODO 需求简化：统计以10分钟为区间的区间内广告点击数量，每次统计最近一分钟的数据，且每10s统计一次
// 比如：12:50 - 12:59，13:20 - 13:29
// 同时，12:51、12:55都算在12:50 - 12:59区间内的数据
// 我们将每个timestamp的分钟都做保留十分位处理即可：12:58 -> 12:50, 12:13 -> 12:10
//转换数据格式，将数据时间转换为仅保留十分位的区间数据
val mapDS = adClickDataDS.map(
data => {
val timestamp = data.timestamp.toLong
//保留分钟的十分位
//操作解释：
// timestamp以ms为单位，/ 1000先将其转换为以s为单位， / 60 转换为以min为单位， / 10取出分钟十分位
// 处理结束后，变为以十分位为位数的时间戳，将其还原为以ms为单位同时仅保留十分位，再 * 10000 * 10 * 60即可
val time = (timestamp / 1000 / 60 / 10) * (1000 * 10 * 60)
//将时间格式化
val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
val day = format.format(new Date(time))
((day, data.ad), 1)
}
)
//按窗口进行统计，每次统计60s数据，每10s统计一次
val resultDS = mapDS.reduceByKeyAndWindow(
(x: Int, y: Int) => x + y, //DS聚合规则
Seconds(60), //窗口大小，每次统计1分钟内的数据
Seconds(10) //窗口步长，每10s滑动，统计一次
)
resultDS.print()
ssc.start()
ssc.awaitTermination()
}
/**
* 广告点击数据的数据模型实体，用于接收并封装Kafka数据
*/
private case class AdClickData(
timestamp: String, //时间戳,
ad: String //广告信息
)
}- 代码解析如下：
a 消费Kafka数据源，封装到数据模型
//创建Kafka消费者
//定义Kafka配置参数
val kafkaPara: Map[String, Object] = Map[String, Object](
//消费者服务器节点
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092,node2:9092,node3:9092",
//消费者组消费主题
ConsumerConfig.GROUP_ID_CONFIG -> "sparkstreamingpractice",
"key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)
//连接Kafka，获取消费数据
val kafkaDataDS = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferBrokers, //采集数据位置的策略，让框架自行匹配如何采集数据
ConsumerStrategies.Subscribe[String, String](Set("sparkstreamingpractice"), kafkaPara) //设置消费者相关参数
)
//将Kafka流数据封装到数据模型中
val adClickDataDS = kafkaDataDS.map(
kafkaData => {
val value = kafkaData.value() //获取数据的值
val splitData = value.split(" ")
//数据格式： timestamp area city userid adid（空格隔开）
AdClickData(splitData(0), splitData(4))
}
)b 将数据进行转换为区间数据
//转换数据格式，将数据时间转换为仅保留十分位的区间数据
val mapDS = adClickDataDS.map(
data => {
val timestamp = data.timestamp.toLong
//保留分钟的十分位
//操作解释：
// timestamp以ms为单位，/ 1000先将其转换为以s为单位， / 60 转换为以min为单位， / 10取出分钟十分位
// 处理结束后，变为以十分位为位数的时间戳，将其还原为以ms为单位同时仅保留十分位，再 * 10000 * 10 * 60即可
val time = timestamp / 1000 / 60 / 10 * 1000 * 10 * 60
//将时间格式化
val format = new SimpleDateFormat("yyyy-MM-dd HH:mm")
val day = format.format(new Date(time))
((day, data.ad), 1)
}
)- 转换规则说明：
- / 操作会默认对结果取整，我们通过这个操作来实现我们的需求。
- 由于timestamp是以ms为单位的系统时间，所以首先需要通过 / 1000转换为以s为单位的时间。然后 / 60，转换为以min为单位的时间。
- 接着我们需要取出分钟的十分位数字，通过 / 10来进行。
- 最后，将时间还原为以ms为单位，之前去除了多少的倍数就还原回去，即 * 1000 * 60 * 10。
c 对区间数据按窗口参数进行聚合
//按窗口进行统计，每次统计60s数据，每10s统计一次
val resultDS = mapDS.reduceByKeyAndWindow(
(x: Int, y: Int) => x + y, //DS聚合规则
Seconds(60), //窗口大小，每次统计1分钟内的数据
Seconds(10) //窗口步长，每10s滑动，统计一次
)注意，指定的窗口大小 & 滑动步长需要为StreamingContext采集周期的整数倍！此案例StreamingContext的设置如下：
（3）代码测试
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
8
0- 0
已为社区贡献4条内容
所有评论(0)