3. 实时风控系统架构：Lambda 架构与 Kappa 架构对比，Spark Streaming 在其中的定位

在贷后实时风险拦截场景中，系统架构的选择直接决定了数据处理的延迟、准确性、运维复杂度以及成本。本章将深入对比两种主流实时架构——Lambda 与 Kappa，并明确 Spark Streaming 在其中的核心定位。

3.1 Lambda 架构：批流分离的经典方案

Lambda 架构由 Nathan Marz 提出，旨在同时满足实时性与全量数据准确性。它将数据流分为两条独立的路径：

• 批处理层（Batch Layer）：存储全量原始数据，定期（如每小时/每天）运行批处理作业，生成准确的离线视图。在贷后场景中，用于计算用户历史逾期率、设备指纹黑名单全量匹配等。
• 速度层（Speed Layer）：处理实时增量数据，提供低延迟（秒级）的近似视图。用于实时交易拦截、当前会话风险评分等。
• 服务层（Serving Layer）：合并批处理视图与实时视图，对外提供统一的查询接口。

3.2 Kappa 架构：流式优先的简化演进

Kappa 架构由 Jay Kreps 提出，核心思想是“一切皆流”。它取消了独立的批处理层，所有数据（包括历史重算）都通过统一的流处理管道处理：

• 统一数据管道：使用 Kafka 等消息队列作为数据主干，所有数据以事件流形式进入。
• 流处理引擎：Spark Streaming / Flink 等引擎实时处理数据，并输出到服务层。
• 历史重算：当逻辑变更时，从 Kafka 的 offset 或对象存储中重新消费全量数据，重新生成结果。

3.3 架构对比：Lambda vs Kappa

3.4 Spark Streaming 在架构中的定位

Spark Streaming 作为 Spark 生态的流处理组件，在两种架构中扮演不同角色：

• 在 Lambda 架构中：Spark Streaming 通常作为速度层的核心引擎，处理秒级微批次数据，输出实时风险评分。同时，Spark SQL / DataFrame API 可用于批处理层，实现代码复用（同一套 DataFrame 转换逻辑可同时用于批和流）。
• 在 Kappa 架构中：Spark Streaming 作为唯一的处理引擎，通过 Structured Streaming 的连续处理模式（Continuous Processing）或微批次模式，统一处理实时数据与历史重算数据。

3.5 贷后场景下的架构选择建议

基于贷后实时风险拦截的业务特性，给出以下选型建议：

• 规则简单、延迟敏感（如实时交易拦截）：优先采用 Kappa 架构 + Spark Structured Streaming，降低运维复杂度，实现秒级响应。
• 需要全量画像与实时拦截结合（如每日更新用户风险等级 + 实时行为评分）：可采用 Lambda 架构，批处理层使用 Spark SQL 生成日级画像，速度层使用 Spark Streaming 处理实时事件。
• 团队技术栈以 Spark 为主：无论选择哪种架构，Spark Streaming 都能提供统一的 API（DataFrame / SQL），减少学习成本。

3.6 代码示例：Spark Streaming 在两种架构中的入口差异

以下代码片段展示了 Spark Streaming 在 Lambda 速度层与 Kappa 统一管道中的典型入口差异：

// ========== Lambda 架构：速度层（Spark Streaming 微批次） ==========
val ssc = new StreamingContext(sparkConf, Seconds(5))
val dStream: DStream[String] = ssc.socketTextStream("localhost", 9999)
dStream
  .map(_.split(","))
  .filter(arr => arr.length == 3)
  .map(arr => (arr(0), arr(1).toDouble))
  .reduceByKey(_ + _)
  .print()
ssc.start()
ssc.awaitTermination()

// ========== Kappa 架构：统一管道（Spark Structured Streaming） ==========
val df = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "risk_events")
  .load()

val riskScores = df
  .selectExpr("CAST(value AS STRING)")
  .as[String]
  .flatMap(parseEvent)  // 自定义解析函数
  .withWatermark("eventTime", "10 minutes")
  .groupBy("userId", window("eventTime", "5 minutes"))
  .agg(avg("score").as("avgRiskScore"))

riskScores
  .writeStream
  .format("console")
  .option("truncate", "false")
  .start()
  .awaitTermination()