4. Spark Streaming 核心原理：DStream、微批次、Receiver 与 Direct 模式

在贷后实时风险拦截场景中，Spark Streaming 是处理高吞吐、低延迟事件流的核心引擎。理解其底层原理——DStream 抽象、微批次调度、Receiver 模式与 Direct 模式——是构建稳定、高效拦截系统的前提。本章从概念到机制，逐一拆解这些核心要素。

4.1 DStream：离散化数据流抽象

DStream（Discretized Stream）是 Spark Streaming 提供的高层抽象，表示一个连续的数据流。从编程模型看，DStream 本质上是一个由 RDD 序列组成的“流”，每个 RDD 对应一个时间窗口内的数据批次。

4.2 微批次架构：时间切片与调度

Spark Streaming 采用 微批次（Micro-Batch） 架构，而非逐条处理。数据按固定时间间隔（如 1 秒、5 秒）切分为批次，每个批次作为一个 Spark 作业提交执行。这种设计带来了吞吐与容错优势，但也引入了固定的调度延迟。

4.3 Receiver 模式：传统可靠接收

Receiver 模式是 Spark Streaming 早期版本（1.x）的主流数据接入方式。每个 Receiver 作为一个长期运行的任务（Task）运行在 Executor 上，负责从数据源（如 Kafka、Flume）拉取数据，并存储为 WAL（Write-Ahead Log）和内存中的 Block。

4.3.1 工作流程

1. Receiver 启动：在 Executor 上启动一个 Receiver 任务，持续从数据源拉取数据。
2. 数据缓冲：Receiver 将接收到的数据切分为 Block，写入 Executor 内存，同时异步写入 WAL（HDFS 或本地文件系统）。
3. Block 汇报：Receiver 向 Driver 的 BlockManager 汇报 Block 元数据。
4. 作业生成：Driver 根据批次间隔，将属于同一批次的 Block 组合为一个 RDD，提交作业。

4.3.2 关键问题

• 数据重复：WAL 写成功但 Block 汇报失败时，Driver 会重放 WAL，导致数据重复消费。
• 背压（Backpressure）：Receiver 拉取速度与处理速度不匹配时，容易导致内存溢出。需手动配置 spark.streaming.backpressure.enabled=true。
• 单点瓶颈：每个 Receiver 是单线程拉取，多分区需启动多个 Receiver。

4.4 Direct 模式：无 Receiver 的精确一次语义

Direct 模式（又称无 Receiver 模式）从 Spark Streaming 1.3 引入，是当前生产环境（尤其是贷后拦截场景）的推荐方案。它不再使用 Receiver 接收数据，而是由 Driver 直接控制 Kafka 分区的偏移量（Offset），每个批次直接读取指定偏移量范围内的数据。

4.4.1 核心机制

• Driver 驱动偏移量：Driver 在每个批次周期查询 Kafka 的最新偏移量，计算出当前批次应读取的起始与结束偏移量。
• Executor 直接读取：生成的 RDD 的每个 Partition 对应一个 Kafka 分区，Executor 直接通过 Kafka 简单消费者 API 拉取数据。
• 偏移量管理：偏移量可以存储在 Zookeeper、Kafka 自身或外部存储（如 MySQL、Redis）中，实现精确一次语义。

4.4.2 与 Receiver 模式对比

4.5 贷后场景选型建议

以下为贷后场景的推荐配置模板：

// Spark Streaming 配置示例（Direct 模式 + 精确一次语义）
val sparkConf = new SparkConf()
  .setAppName("PostLoanRiskInterceptor")
  .set("spark.streaming.kafka.maxRatePerPartition", "1000")   // 每分区每秒最大拉取条数
  .set("spark.streaming.backpressure.enabled", "true")        // 启用背压
  .set("spark.streaming.backpressure.initialRate", "500")     // 初始速率
  .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

val ssc = new StreamingContext(sparkConf, Seconds(2))          // 2秒微批次

// Direct 模式创建流
val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> "kafka1:9092,kafka2:9092",
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[StringDeserializer],
  "auto.offset.reset" -> "latest",
  "enable.auto.commit" -> (false: java.lang.Boolean)         // 手动管理偏移量
)

val topics = Array("loan_repay_events", "overdue_events")
val stream = KafkaUtils.createDirectStream[String, String](
  ssc,
  LocationStrategies.PreferConsistent,
  ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
)