4. 批处理基础：Spark SQL 与 DataSet API、批处理 ETL 流程、与实时计算的异同

在风控系统中，批处理承担着“历史回溯、模型训练、离线报表、批量名单加工”等核心任务。本章将深入讲解 Spark SQL 与 DataSet API 在批处理场景下的应用，梳理标准批处理 ETL 流程，并系统对比批处理与实时计算在架构、延迟、数据语义上的本质差异。

4.1 Spark SQL 与 DataSet API 核心概念

Spark SQL 是 Spark 生态中用于结构化数据处理的模块，它提供了 DataFrame 和 DataSet 两种高级抽象。在风控批处理中，我们主要使用 DataSet API 来获得编译时类型安全与强类型检查。

4.2 批处理 ETL 标准流程

一个典型的风控批处理 ETL 作业通常包含以下五个阶段。每个阶段都有明确的输入输出和异常处理策略。

1. 数据源接入（Source）：从 HDFS、Hive 表、Kafka（批量消费）或 JDBC 读取原始数据，如交易流水、用户注册日志。
2. 数据清洗（Clean）：过滤无效字段、处理空值、去重、格式标准化。例如将时间戳统一为 UTC+8。
3. 特征加工（Transform）：基于业务规则进行聚合、关联、窗口计算。例如计算用户过去 7 天交易总金额、设备指纹聚类。
4. 规则匹配（Match）：将加工后的特征与风控规则库（如决策树、名单库）进行匹配，输出风险等级与处置建议。
5. 结果输出（Sink）：将结果写入 Hive 分区表、MySQL 或 ES，供后续报表、模型训练或实时缓存使用。

4.3 批处理与实时计算的异同

在风控系统中，批处理和实时计算并非互斥，而是互补。下表从多个维度对比两者的核心差异，帮助你在架构选型时做出正确判断。

4.4 批处理代码示例：DataSet API 实现 ETL

以下代码展示了一个典型的风控批处理 ETL 作业，使用 Spark DataSet API 进行强类型转换。

import org.apache.spark.sql.{SparkSession, Encoders}
import org.apache.spark.sql.functions._

// 定义强类型 case class
case class RawTransaction(
  transId: String,
  userId: String,
  amount: Double,
  transTime: String,
  deviceId: String
)

case class RiskFeature(
  userId: String,
  totalAmount7d: Double,
  transCount7d: Long,
  deviceCount7d: Long
)

val spark = SparkSession.builder()
  .appName("RiskBatchETL")
  .enableHiveSupport()
  .getOrCreate()

import spark.implicits._

// 1. 读取原始数据（Hive 表）
val rawDS: Dataset[RawTransaction] = spark
  .table("ods.transaction_log")
  .as[RawTransaction]

// 2. 数据清洗：过滤异常金额、标准化时间
val cleanedDS = rawDS
  .filter(_.amount > 0)
  .filter(_.transId.nonEmpty)
  .map(t => t.copy(transTime = normalizeTime(t.transTime)))

// 3. 特征加工：过去7天聚合
val riskFeatureDS: Dataset[RiskFeature] = cleanedDS
  .groupByKey(_.userId)
  .agg(
    sum("amount").as[Double],
    count("transId").as[Long],
    countDistinct("deviceId").as[Long]
  )
  .map { case (userId, totalAmt, cnt, devCnt) =>
    RiskFeature(userId, totalAmt, cnt, devCnt)
  }

// 4. 规则匹配（示例：金额超过阈值）
val highRiskDS = riskFeatureDS.filter(_.totalAmount7d > 100000)

// 5. 结果输出到 Hive 分区表
highRiskDS
  .toDF()
  .write
  .mode("overwrite")
  .format("parquet")
  .partitionBy("dt")
  .saveAsTable("dwd.risk_user_7d")

4.5 批流一体架构下的协同策略

在批流一体风控系统中，批处理与实时计算并非孤立运行，而是通过以下机制协同工作：

• 数据同源：批处理和实时计算从同一 Kafka Topic 或 Hive 表读取原始数据，保证特征一致性。
• 特征互补：实时计算产出秒级特征（如当前会话行为），批处理产出历史统计特征（如 30 天累计金额），两者在决策服务层融合。
• 模型同步：批处理负责模型训练和离线评估，将模型参数（如阈值、权重）写入共享存储（Redis / MySQL），实时计算在运行时加载最新模型。
• 结果校准：批处理每日对实时决策结果进行回溯校验，发现偏差后触发告警或自动修正规则。