大数据技术之Spark Streaming

摘要：Spark Streaming 是 Apache Spark 中用于实时流数据处理的模块。本文将深入剖析 Spark Streaming 的核心概念、DStream 抽象、数据源集成、转换操作、窗口计算以及企业级实战案例，帮助你构建高吞吐、低延迟、容错的实时数据处理系统。

一、Spark Streaming 概述

1.1 什么是 Spark Streaming

Spark Streaming 是 Spark 用于流式数据（streaming data）处理的模块。它使得构建可扩展的、容错的流应用程序变得更加容易。

Spark Streaming 支持的数据输入源非常丰富：

• 消息队列：Kafka、Flume、Kinesis
• 文件系统：HDFS、S3
• 网络数据源：TCP 套接字、Twitter 等

数据输入后，可以用 Spark 的高度抽象原语如 map、reduce、join、window 等进行运算。结果也能保存在 HDFS、数据库、Dashboard 等多种输出端。

1.2 Spark Streaming 的核心特点

1.2.1 易用性

Spark Streaming 将 Spark 的语言集成 API 带到流处理领域，让你可以用写批处理作业相同的方式编写流处理作业。支持 Java、Scala 和 Python。

1.2.2 容错性

Spark Streaming 能够自动恢复丢失的工作和算子状态（例如滑动窗口），无需额外代码。

1.2.3 易整合到 Spark 体系

通过运行在 Spark 上，Spark Streaming 让你可以重用相同的代码进行批处理，将流与历史数据 Join，或对流状态运行 ad-hoc 查询。

1.3 Spark Streaming 架构

1.3.1 整体架构图

Spark Streaming 的整体架构由 Spark Streaming Driver 和 Executor 两部分组成：

Driver 端组件：

• Receiver Tracker：跟踪 Receiver 的状态
• Job Generator：根据时间间隔生成 JobSet
• Job Scheduler：调度并运行 Job
• SparkContext：底层 Spark 计算引擎

Executor 端组件：

• Receiver：接收数据流
• Block Manager：管理接收到的数据块，并复制到另一个 BM 实现容错

1.3.2 背压机制（Backpressure）

Spark 1.5 以前，用户需要通过静态配置参数 spark.streaming.receiver.maxRate 来限制 Receiver 的数据接收速率。这种方式虽然可以防止内存溢出，但会导致资源利用率下降。

Spark 1.5 开始引入了背压机制：根据 JobScheduler 反馈的作业执行信息来动态调整 Receiver 数据接收率。

# 启用背压机制
spark.streaming.backpressure.enabled=true

二、DStream 入门

2.1 DStream 是什么

**DStream（Discretized Stream）**是 Spark Streaming 的基础抽象，代表持续性的数据流和经过各种 Spark 原语操作后的结果数据流。

在内部实现上，DStream 是一系列连续的 RDD 来表示。每个 RDD 含有一段时间间隔内的数据。

核心概念：

• 时间片（Batch Interval）：将连续的数据流切分成离散的小批次
• 每个批次对应一个 RDD
• 对 DStream 的操作实际上是对每个 RDD 的操作

2.2 WordCount 案例实操

2.2.1 添加依赖

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.12</artifactId>
<version>3.0.0</version>
</dependency>

2.2.2 编写代码

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object StreamWordCount {
def main(args: Array[String]): Unit = {
// 1. 初始化 Spark 配置信息
val sparkConf = new SparkConf()
.setMaster("local[*]")
.setAppName("StreamWordCount")

// 2. 初始化 SparkStreamingContext，批次间隔 3 秒
val ssc = new StreamingContext(sparkConf, Seconds(3))

// 3. 通过监控端口创建 DStream
val lineStreams = ssc.socketTextStream("localhost", 9999)

// 4. 将每一行数据做切分，形成一个个单词
val wordStreams = lineStreams.flatMap(_.split(" "))

// 5. 将单词映射成元组 (word, 1)
val wordAndOneStreams = wordStreams.map((_, 1))

// 6. 将相同的单词次数做统计
val wordAndCountStreams = wordAndOneStreams.reduceByKey(_ + _)

// 7. 打印
wordAndCountStreams.print()

// 8. 启动 SparkStreamingContext
ssc.start()
ssc.awaitTermination()
}
}

2.2.3 启动程序并发送数据

# 终端1：启动 netcat 发送数据
nc -lk 9999

# 输入内容
hello spark
hello scala

2.3 DStream 转换操作解析

对数据的操作也是按照 RDD 为单位来进行的：

计算过程由 Spark Engine 来完成：

• 输入数据流被切分成批次
• Spark Streaming 将批次交给 Spark Engine 处理
• 输出处理后的批次结果

三、DStream 创建

3.1 RDD 队列

测试过程中，可以通过 ssc.queueStream(queueOfRDDs) 来创建 DStream。每一个推送到队列中的 RDD，都会作为一个 DStream 处理。

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.rdd.RDD
import scala.collection.mutable

object RDDStream {
def main(args: Array[String]): Unit = {
// 1. 初始化 Spark 配置
val conf = new SparkConf().setMaster("local[*]").setAppName("RDDStream")

// 2. 初始化 StreamingContext，批次间隔 4 秒
val ssc = new StreamingContext(conf, Seconds(4))

// 3. 创建 RDD 队列
val rddQueue = new mutable.Queue[RDD[Int]]()

// 4. 创建 QueueInputDStream
val inputStream = ssc.queueStream(rddQueue, oneAtATime = false)

// 5. 处理队列中的 RDD 数据
val mappedStream = inputStream.map((_, 1))
val reducedStream = mappedStream.reduceByKey(_ + _)

// 6. 打印结果
reducedStream.print()

// 7. 启动任务
ssc.start()

// 8. 循环创建并向 RDD 队列中放入 RDD
for (i <- 1 to 5) {
rddQueue += ssc.sparkContext.makeRDD(1 to 300, 10)
Thread.sleep(2000)
}

ssc.awaitTermination()
}
}

3.2 自定义数据源

需要继承 Receiver，并实现 onStart、onStop 方法来自定义数据源采集。

import java.io.{BufferedReader, InputStreamReader}
import java.net.Socket
import java.nio.charset.StandardCharsets
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.receiver.Receiver

class CustomerReceiver(host: String, port: Int)
extends Receiver[String](StorageLevel.MEMORY_ONLY) {

// 最初启动时调用，读数据并发送给 Spark
override def onStart(): Unit = {
new Thread("Socket Receiver") {
override def run() {
receive()
}
}.start()
}

// 读数据并发送给 Spark
def receive(): Unit = {
var socket: Socket = new Socket(host, port)
var input: String = null
val reader = new BufferedReader(
new InputStreamReader(socket.getInputStream, StandardCharsets.UTF_8)
)

input = reader.readLine()
while (!isStopped() && input != null) {
store(input)
input = reader.readLine()
}

reader.close()
socket.close()
restart("restart")
}

override def onStop(): Unit = {}
}

使用自定义数据源：

val lineStream = ssc.receiverStream(new CustomerReceiver("localhost", 9999))
val wordStream = lineStream.flatMap(_.split(" "))
val wordAndOneStream = wordStream.map((_, 1))
val wordAndCountStream = wordAndOneStream.reduceByKey(_ + _)
wordAndCountStream.print()

3.3 Kafka 数据源（面试、开发重点）

3.3.1 版本选型说明

3.3.2 Kafka 0-10 Direct 模式（推荐）

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object DirectAPI {
def main(args: Array[String]): Unit = {
// 1. 创建 SparkConf
val sparkConf = new SparkConf()
.setAppName("KafkaDirect")
.setMaster("local[*]")

// 2. 创建 StreamingContext
val ssc = new StreamingContext(sparkConf, Seconds(3))

// 3. 定义 Kafka 参数
val kafkaPara: Map[String, Object] = Map[String, Object](
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "localhost:9092",
ConsumerConfig.GROUP_ID_CONFIG -> "atguigu",
"key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)

// 4. 读取 Kafka 数据创建 DStream
val kafkaDStream: InputDStream[ConsumerRecord[String, String]] =
KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferConsistent,
ConsumerStrategies.Subscribe[String, String](Set("atguigu"), kafkaPara)
)

// 5. 将每条消息的 value 取出
val valueDStream: DStream[String] = kafkaDStream.map(record => record.value())

// 6. 计算 WordCount
valueDStream.flatMap(_.split(" "))
.map((_, 1))
.reduceByKey(_ + _)
.print()

// 7. 开启任务
ssc.start()
ssc.awaitTermination()
}
}

关键配置说明：

3.3.3 手动维护 Offset

import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils, OffsetRange}

object DirectAPIHandler {
def main(args: Array[String]): Unit = {
val ssc = new StreamingContext(sparkConf, Seconds(3))

val kafkaPara = Map[String, String](
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "localhost:9092",
ConsumerConfig.GROUP_ID_CONFIG -> "atguigu"
)

// 获取上一次启动最后保留的 Offset（可从 MySQL/Redis/ZK 获取）
val fromOffsets: Map[TopicAndPartition, Long] = Map(
TopicAndPartition("atguigu", 0) -> 20
)

// 读取 Kafka 数据
val kafkaDStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, String](
ssc, kafkaPara, fromOffsets,
(m: MessageAndMetadata[String, String]) => m.message()
)

// 获取当前消费数据的 offset 信息
var offsetRanges = Array.empty[OffsetRange]

val wordToCountDStream = kafkaDStream.transform { rdd =>
offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
rdd
}.flatMap(_.split(" "))
.map((_, 1))
.reduceByKey(_ + _)

// 打印 Offset 信息并保存到外部存储
wordToCountDStream.foreachRDD(rdd => {
for (o <- offsetRanges) {
println(s"${o.topic}:${o.partition}:${o.fromOffset}:${o.untilOffset}")
// 保存到 MySQL/Redis/ZK
}
rdd.foreach(println)
})

ssc.start()
ssc.awaitTermination()
}
}

四、DStream 转换

DStream 上的操作与 RDD 类似，分为 **Transformations（转换）**和 **Output Operations（输出）**两种。

4.1 无状态转化操作

无状态转化操作就是把简单的 RDD 转化操作应用到每个批次上，也就是转化 DStream 中的每一个 RDD。

注意：reduceByKey() 会归约每个时间区间中的数据，但不会归约不同区间之间的数据。

4.2 Transform 操作

Transform 允许在 DStream 上执行任意的 RDD-to-RDD 函数，方便扩展 Spark API。

val lineDStream: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

// 转换为 RDD 操作
val wordAndCountDStream = lineDStream.transform { rdd =>
val words: RDD[String] = rdd.flatMap(_.split(" "))
val value: RDD[(String, Int)] = words.map((_, 1))
value.reduceByKey(_ + _)
}

4.3 Join 操作

两个流之间的 Join 需要两个流的批次大小一致，这样才能同时触发计算。

val lineDStream1 = ssc.socketTextStream("localhost", 9999)
val lineDStream2 = ssc.socketTextStream("localhost", 8888)

// 转换为 KV 类型
val wordToOneDStream = lineDStream1.flatMap(_.split(" ")).map((_, 1))
val wordToADStream = lineDStream2.flatMap(_.split(" ")).map((_, "a"))

// 流的 JOIN
val joinDStream = wordToOneDStream.join(wordToADStream)
joinDStream.print()

4.4 有状态转化操作

4.4.1 UpdateStateByKey

用于跨批次维护状态，记录历史数据。

object WorldCount {
def main(args: Array[String]): Unit = {
// 定义更新状态方法
val updateFunc = (values: Seq[Int], state: Option[Int]) => {
val currentCount = values.foldLeft(0)(_ + _)
val previousCount = state.getOrElse(0)
Some(currentCount + previousCount)
}

val conf = new SparkConf().setMaster("local[*]").setAppName("NetworkWordCount")
val ssc = new StreamingContext(conf, Seconds(3))

// 必须设置检查点目录
ssc.checkpoint("./ck")

val lines = ssc.socketTextStream("localhost", 9999)
val words = lines.flatMap(_.split(" "))
val pairs = words.map((_, 1))

// 使用 updateStateByKey 更新状态，统计从运行开始以来的总次数
val stateDstream = pairs.updateStateByKey[Int](updateFunc)
stateDstream.print()

ssc.start()
ssc.awaitTermination()
}
}

输出示例：

——————————————-
Time: 1504685181000 ms
——————————————-
(shi,1)
(shui,1)
(ni,1)

——————————————-
Time: 1504685187000 ms
——————————————-
(shi,1)
(ma,1)
(hao,1)
(shui,1)

4.4.2 Window Operations

Window Operations 可以设置窗口大小和滑动步长来动态获取当前 Streaming 的状态。

注意：两者都必须是采集周期大小的整数倍。

object WindowWordCount {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[2]").setAppName("WindowWordCount")
val ssc = new StreamingContext(conf, Seconds(3))
ssc.checkpoint("./ck")

val lines = ssc.socketTextStream("localhost", 9999)
val words = lines.flatMap(_.split(" "))
val pairs = words.map((_, 1))

// 3秒一个批次，窗口12秒，滑步6秒
val wordCounts = pairs.reduceByKeyAndWindow(
(a: Int, b: Int) => (a + b),
Seconds(12),
Seconds(6)
)

wordCounts.print()
ssc.start()
ssc.awaitTermination()
}
}

4.4.3 窗口操作方法汇总

增量计算示例：

val ipDStream = accessLogsDStream.map(logEntry => (logEntry.getIpAddress(), 1))

val ipCountDStream = ipDStream.reduceByKeyAndWindow(
(x, y) => x + y, // 加上新进入窗口的批次元素
(x, y) => x – y, // 移除离开窗口的老批次元素
Seconds(30), // 窗口时长
Seconds(10) // 滑动步长
)

五、DStream 输出

输出操作指定了对流数据经转化操作得到的数据所要执行的操作。

foreachRDD 最佳实践

// 错误做法1：连接写在 driver 层面（无法序列化）
// 错误做法2：写在 foreach 中，每条数据都创建连接

// 正确做法：使用 foreachPartition，在分区级别创建连接
wordCounts.foreachRDD(rdd => {
rdd.foreachPartition(partitionOfRecords => {
// 每个分区创建一个连接
val connection = createNewConnection()
partitionOfRecords.foreach(record => {
connection.write(record)
})
connection.close()
})
})

六、优雅关闭

流式任务需要 7×24 小时执行，但涉及升级代码时需要主动停止程序。分布式程序无法逐个进程杀死，因此优雅关闭至关重要。

6.1 实现方案

使用外部文件系统（如 HDFS）来控制内部程序关闭：

import java.net.URI
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.streaming.{StreamingContext, StreamingContextState}

class MonitorStop(ssc: StreamingContext) extends Runnable {
override def run(): Unit = {
val fs = FileSystem.get(new URI("hdfs://localhost:9000"), new Configuration(), "root")

while (true) {
Thread.sleep(5000)
val state = ssc.getState
val bool = fs.exists(new Path("hdfs://localhost:9000/stopSpark"))

if (bool && state == StreamingContextState.ACTIVE) {
ssc.stop(stopSparkContext = true, stopGracefully = true)
System.exit(0)
}
}
}
}

6.2 配置优雅关闭

val sparkConf = new SparkConf()
.setMaster("local[*]")
.setAppName("SparkTest")
// 设置优雅的关闭
.set("spark.streaming.stopGracefullyOnShutdown", "true")

val ssc = new StreamingContext(sparkConf, Seconds(5))
ssc.checkpoint("./ck")

// 启动监控线程
new Thread(new MonitorStop(ssc)).start()

ssc.start()
ssc.awaitTermination()

七、实战项目：广告实时统计系统

7.1 项目概述

构建一个广告点击实时统计系统，包含三个核心需求：

7.2 环境准备

7.2.1 POM 依赖

<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid</artifactId>
<version>1.1.10</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.27</version>
</dependency>
</dependencies>

7.2.2 MySQL 建表

— 黑名单表
CREATE TABLE black_list (
userid CHAR(1) PRIMARY KEY
);

— 单日用户广告点击次数表
CREATE TABLE user_ad_count (
dt VARCHAR(255),
userid CHAR(1),
adid CHAR(1),
count BIGINT,
PRIMARY KEY (dt, userid, adid)
);

— 每天各大区各城市广告点击总数表
CREATE TABLE area_city_ad_count (
dt VARCHAR(255),
area VARCHAR(255),
city VARCHAR(255),
adid VARCHAR(255),
count BIGINT,
PRIMARY KEY (dt, area, city, adid)
);

7.3 工具类封装

7.3.1 Kafka 工具类

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}

object MyKafkaUtil {
private val properties = PropertiesUtil.load("config.properties")
val broker_list: String = properties.getProperty("kafka.broker.list")

val kafkaParam = Map(
"bootstrap.servers" -> broker_list,
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> "commerce-consumer-group",
"auto.offset.reset" -> "latest",
"enable.auto.commit" -> (true: java.lang.Boolean)
)

def getKafkaStream(topic: String, ssc: StreamingContext):
InputDStream[ConsumerRecord[String, String]] = {
KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferConsistent,
ConsumerStrategies.Subscribe[String, String](Array(topic), kafkaParam)
)
}
}

7.3.2 JDBC 工具类

import java.sql.{Connection, PreparedStatement, ResultSet}
import java.util.Properties
import com.alibaba.druid.pool.DruidDataSource

object JdbcUtil {
var dataSource: DruidDataSource = init()

def init(): DruidDataSource = {
val properties = new Properties()
val config = PropertiesUtil.load("config.properties")
val ds = new DruidDataSource()
ds.setUrl(config.getProperty("jdbc.url"))
ds.setUsername(config.getProperty("jdbc.user"))
ds.setPassword(config.getProperty("jdbc.password"))
ds.setInitialSize(config.getProperty("jdbc.datasource.size").toInt)
ds
}

def getConnection: Connection = dataSource.getConnection

// 单条数据插入/更新
def executeUpdate(connection: Connection, sql: String, params: Array[Any]): Int = {
var rtn = 0
var pstmt: PreparedStatement = null
try {
connection.setAutoCommit(false)
pstmt = connection.prepareStatement(sql)
if (params != null && params.length > 0) {
for (i <- params.indices) {
pstmt.setObject(i + 1, params(i))
}
}
rtn = pstmt.executeUpdate()
connection.commit()
pstmt.close()
} catch {
case e: Exception => e.printStackTrace()
}
rtn
}

// 批量数据插入
def executeBatchUpdate(connection: Connection, sql: String,
paramsList: Iterable[Array[Any]]): Array[Int] = {
var rtn: Array[Int] = null
var pstmt: PreparedStatement = null
try {
connection.setAutoCommit(false)
pstmt = connection.prepareStatement(sql)
for (params <- paramsList) {
if (params != null && params.length > 0) {
for (i <- params.indices) {
pstmt.setObject(i + 1, params(i))
}
pstmt.addBatch()
}
}
rtn = pstmt.executeBatch()
connection.commit()
pstmt.close()
} catch {
case e: Exception => e.printStackTrace()
}
rtn
}

// 判断数据是否存在
def isExist(connection: Connection, sql: String, params: Array[Any]): Boolean = {
var flag = false
var pstmt: PreparedStatement = null
try {
pstmt = connection.prepareStatement(sql)
for (i <- params.indices) {
pstmt.setObject(i + 1, params(i))
}
flag = pstmt.executeQuery().next()
pstmt.close()
} catch {
case e: Exception => e.printStackTrace()
}
flag
}

// 获取单条数据
def getDataFromMysql(connection: Connection, sql: String,
params: Array[Any]): Long = {
var result: Long = 0L
var pstmt: PreparedStatement = null
try {
pstmt = connection.prepareStatement(sql)
for (i <- params.indices) {
pstmt.setObject(i + 1, params(i))
}
val resultSet = pstmt.executeQuery()
while (resultSet.next()) {
result = resultSet.getLong(1)
}
resultSet.close()
pstmt.close()
} catch {
case e: Exception => e.printStackTrace()
}
result
}
}

7.4 需求一：广告黑名单

7.4.1 需求描述

实现实时的动态黑名单机制：将每天对某个广告点击超过 100 次的用户拉黑。

7.4.2 思路分析

7.4.3 代码实现

case class Ads_log(timestamp: Long, area: String, city: String, userid: String, adid: String)

object BlackListHandler {
private val sdf = new SimpleDateFormat("yyyy-MM-dd")

// 添加黑名单
def addBlackList(filterAdsLogDSteam: DStream[Ads_log]): Unit = {
// 1. 将数据转换结构：ads_log => ((date, user, adid), 1)
val dateUserAdToOne = filterAdsLogDSteam.map(adsLog => {
val date = sdf.format(new Date(adsLog.timestamp))
((date, adsLog.userid, adsLog.adid), 1L)
})

// 2. 统计单日每个用户点击每个广告的总次数
val dateUserAdToCount = dateUserAdToOne.reduceByKey(_ + _)

dateUserAdToCount.foreachRDD(rdd => {
rdd.foreachPartition(iter => {
val connection = JdbcUtil.getConnection
iter.foreach { case ((dt, user, ad), count) =>
// 插入或更新点击次数
JdbcUtil.executeUpdate(connection,
"INSERT INTO user_ad_count (dt, userid, adid, count) VALUES (?, ?, ?, ?) ON DUPLICATE KEY UPDATE count = count + ?",
Array(dt, user, ad, count, count))

// 查询更新后的总次数
val ct = JdbcUtil.getDataFromMysql(connection,
"SELECT count FROM user_ad_count WHERE dt = ? AND userid = ? AND adid = ?",
Array(dt, user, ad))

// 超过阈值则加入黑名单
if (ct >= 100) {
JdbcUtil.executeUpdate(connection,
"INSERT INTO black_list (userid) VALUES (?) ON DUPLICATE KEY UPDATE userid = ?",
Array(user, user))
}
}
connection.close()
})
})
}

// 根据黑名单过滤数据
def filterByBlackList(adsLogDStream: DStream[Ads_log]): DStream[Ads_log] = {
adsLogDStream.transform(rdd => {
rdd.filter(adsLog => {
val connection = JdbcUtil.getConnection
val bool = JdbcUtil.isExist(connection,
"SELECT * FROM black_list WHERE userid = ?",
Array(adsLog.userid))
connection.close()
!bool
})
})
}
}

7.5 需求二：广告点击量实时统计

7.5.1 需求描述

实时统计每天各地区各城市各广告的点击总流量，并存入 MySQL。

7.5.2 代码实现

object DateAreaCityAdCountHandler {
private val sdf = new SimpleDateFormat("yyyy-MM-dd")

def saveDateAreaCityAdCountToMysql(filterAdsLogDStream: DStream[Ads_log]): Unit = {
// 1. 统计每天各大区各个城市广告点击总数
val dateAreaCityAdToCount = filterAdsLogDStream.map(ads_log => {
val timestamp = ads_log.timestamp
val dt = sdf.format(new Date(timestamp))
((dt, ads_log.area, ads_log.city, ads_log.adid), 1L)
}).reduceByKey(_ + _)

// 2. 将单个批次统计后的数据与 MySQL 数据合并更新
dateAreaCityAdToCount.foreachRDD(rdd => {
rdd.foreachPartition(iter => {
val connection = JdbcUtil.getConnection
iter.foreach { case ((dt, area, city, adid), count) =>
JdbcUtil.executeUpdate(connection,
"INSERT INTO area_city_ad_count (dt, area, city, adid, count) VALUES (?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE count = count + ?",
Array(dt, area, city, adid, count, count))
}
connection.close()
})
})
}
}

7.6 需求三：最近一小时广告点击量

7.6.1 需求描述

统计最近一小时（窗口 1 小时）广告分时点击总数。

输出示例：

1: List [15:50 -> 10, 15:51 -> 25, 15:52 -> 30]
2: List [15:50 -> 10, 15:51 -> 25, 15:52 -> 30]
3: List [15:50 -> 10, 15:51 -> 25, 15:52 -> 30]

7.6.2 代码实现

object LastHourAdCountHandler {
private val sdf = new SimpleDateFormat("HH:mm")

def getAdHourMintToCount(filterAdsLogDStream: DStream[Ads_log]):
DStream[(String, List[(String, Long)])] = {

// 1. 开窗，时间间隔为 1 个小时
val windowAdsLogDStream = filterAdsLogDStream.window(Minutes(60))

// 2. 转换数据结构：ads_log => ((adid, hm), 1L)
val adHmToOneDStream = windowAdsLogDStream.map(adsLog => {
val timestamp = adsLog.timestamp
val hm = sdf.format(new Date(timestamp))
((adsLog.adid, hm), 1L)
})

// 3. 统计总数：((adid, hm), 1L) => ((adid, hm), sum)
val adHmToCountDStream = adHmToOneDStream.reduceByKey(_ + _)

// 4. 转换数据结构：((adid, hm), sum) => (adid, (hm, sum))
val adToHmCountDStream = adHmToCountDStream.map {
case ((adid, hm), count) => (adid, (hm, count))
}

// 5. 按照 adid 分组，组内按照时分排序
adToHmCountDStream.groupByKey().mapValues(iter =>
iter.toList.sortWith((t1, t2) => t1._1 < t2._1)
)
}
}

7.7 主程序整合

object RealTimeApp {
def main(args: Array[String]): Unit = {
// 1. 创建 SparkConf
val sparkConf = new SparkConf()
.setMaster("local[*]")
.setAppName("RealTimeApp")

// 2. 创建 StreamingContext
val ssc = new StreamingContext(sparkConf, Seconds(3))

// 3. 读取 Kafka 数据
val topic = PropertiesUtil.load("config.properties").getProperty("kafka.topic")
val kafkaDStream = MyKafkaUtil.getKafkaStream(topic, ssc)

// 4. 将每一行数据转换为样例类对象
val adsLogDStream = kafkaDStream.map(record => {
val arr = record.value().split(" ")
Ads_log(arr(0).toLong, arr(1), arr(2), arr(3), arr(4))
})

// 5. 根据 MySQL 中的黑名单表进行数据过滤
val filterAdsLogDStream = BlackListHandler.filterByBlackList(adsLogDStream)

// 6. 将满足要求的用户写入黑名单
BlackListHandler.addBlackList(filterAdsLogDStream)

// 7. 统计每天各大区各个城市广告点击总数并保存至 MySQL
DateAreaCityAdCountHandler.saveDateAreaCityAdCountToMysql(filterAdsLogDStream)

// 8. 统计最近一小时广告分时点击总数
val adToHmCountListDStream = LastHourAdCountHandler.getAdHourMintToCount(filterAdsLogDStream)
adToHmCountListDStream.print()

// 9. 启动任务
ssc.start()
ssc.awaitTermination()
}
}

八、总结

Spark Streaming 通过将实时流数据切分为小批次（Micro-batch）来处理，兼顾了实时性和批处理的稳定性。在实际生产环境中，Kafka + Spark Streaming + MySQL/Redis 是最常见的实时数据处理架构组合。

随着 Spark 的发展，Structured Streaming 基于 DataFrame/DataSet API 提供了更高级的流处理能力，但 DStream 仍是理解 Spark 流处理原理的重要基础。