4. 原始数据接入：从 Hive/MySQL/Kafka 批量读取风控原始事件数据

风控特征工程的第一步是原始数据接入。风控系统需要处理的原始事件数据（如用户注册、登录、交易流水、设备指纹）通常分布在不同的存储系统中。本章聚焦于如何使用 Spark 批处理引擎，从三种最主流的数据源——Hive、MySQL 和 Kafka——高效、可靠地批量读取数据，为后续的特征分层加工奠定基础。

4.1 数据源特性对比与选型

不同的数据源在存储格式、访问模式、数据量级和延迟要求上存在显著差异。选择合适的读取策略是性能优化的第一步。

4.2 Hive 批量读取：分区裁剪与列式优化

Hive 是风控离线特征计算的核心数据湖。Spark 通过 Spark SQL 直接读取 Hive 表，天然支持分区发现和谓词下推。

4.2.1 标准读取模板

// 1. 初始化 SparkSession 并启用 Hive 支持
val spark = SparkSession.builder()
  .appName("RiskFeature_HiveReader")
  .config("spark.sql.adaptive.enabled", "true")
  .enableHiveSupport()
  .getOrCreate()

// 2. 读取 Hive 分区表，指定分区过滤
val rawTradeDF = spark.sql(
  """
    SELECT
      user_id,
      trans_id,
      amount,
      trans_time,
      merchant_id,
      device_id
    FROM risk_db.trade_records
    WHERE dt = '2025-03-01'
      AND hour BETWEEN 10 AND 18
      AND trans_type = 'PAY'
  """
)

// 3. 缓存中间结果（仅当后续多次复用）
rawTradeDF.cache()
rawTradeDF.count()  // 触发缓存

4.2.2 分区裁剪原理

下图展示了 Spark 读取 Hive 表时的分区过滤与列裁剪流程。Spark SQL 优化器会将 WHERE 条件中的分区字段下推到 Hive Metastore，只读取匹配的分区目录，大幅减少扫描数据量。

4.3 MySQL 批量读取：JDBC 分区策略与连接池

MySQL 通常存储风控系统的元数据、用户标签和实时配置表。Spark 通过 JDBC 连接器读取 MySQL，但直接读取大表容易导致数据库连接超时或内存溢出。

4.3.1 分区读取配置

通过指定 partitionColumn、lowerBound、upperBound 和 numPartitions，Spark 可以将一个大的 JDBC 查询拆分为多个并行子查询，每个子查询读取一个数据分片。

val mysqlDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql://risk-mysql:3306/risk_db?useSSL=false")
  .option("dbtable", "user_blacklist")
  .option("user", "risk_app")
  .option("password", "encrypted_password")
  // 分区键：使用自增主键 id
  .option("partitionColumn", "id")
  .option("lowerBound", 1)
  .option("upperBound", 50000000)
  .option("numPartitions", 20)
  // 每次拉取 5000 行，避免 OOM
  .option("fetchSize", 5000)
  .option("connectionTimeout", "30000")
  .load()

// 打印分区数验证
println(s"MySQL 读取分区数: ${mysqlDF.rdd.getNumPartitions}")

4.3.2 连接池与重试机制

生产环境中，MySQL 实例可能因负载波动导致连接失败。建议在 Spark 作业级别配置连接池参数和重试策略。

4.4 Kafka 批量读取：Offset 管理与回溯消费

虽然 Kafka 是流式数据源，但在风控批处理场景中，经常需要回溯过去 N 小时或某天的数据用于特征回填或模型训练。Spark 提供了两种批量读取 Kafka 的方式。

4.4.1 基于 Offset 范围的批量消费

使用 kafka 数据源，通过 startingOffsets 和 endingOffsets 指定精确的 offset 范围，实现批处理语义。

val kafkaBatchDF = spark.read
  .format("kafka")
  .option("kafka.bootstrap.servers", "kafka-cluster:9092")
  .option("subscribe", "risk_raw_events")
  // 指定精确 offset 范围（JSON 格式）
  .option("startingOffsets", """{"risk_raw_events":{"0":1000,"1":2000,"2":1500}}""")
  .option("endingOffsets", """{"risk_raw_events":{"0":5000,"1":6000,"2":4500}}""")
  .option("failOnDataLoss", "false")
  .load()

// 解析 Kafka 消息的 value（通常为 JSON 或 Avro）
import org.apache.spark.sql.functions._

val parsedDF = kafkaBatchDF
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "partition", "offset", "timestamp")
  .withColumn("event_data", from_json(col("value"), schema))  // schema 为预定义 StructType
  .select("event_data.*")

4.4.2 基于时间戳的回溯读取

如果不知道精确的 offset，可以使用 Kafka 0.10+ 的时间戳索引功能，按时间范围定位 offset。

// 按时间戳回溯：读取过去 6 小时的数据
val sixHoursAgo = System.currentTimeMillis() - 6 * 60 * 60 * 1000L

val kafkaTimeBasedDF = spark.read
  .format("kafka")
  .option("kafka.bootstrap.servers", "kafka-cluster:9092")
  .option("subscribe", "risk_raw_events")
  .option("startingOffsets", s"""{"risk_raw_events":{"0":$sixHoursAgo}}""")
  .option("endingOffsets", """{"risk_raw_events":{"0":-1}}""")  // -1 表示最新
  .load()

4.5 多源数据统一接入架构

在实际风控系统中，一个特征计算作业往往需要同时读取多个数据源。下图展示了一个典型的批处理接入架构，数据经过统一 Schema 校验后进入特征加工层。

4.6 接入层性能调优清单

以下是原始数据接入阶段的关键调优项，建议在每次作业上线前逐项检查。

• Hive 调优： 确认分区过滤条件已生效（通过 EXPLAIN 查看 Physical Plan）；使用 ORC/Parquet 列式存储；开启 spark.sql.hive.convertMetastoreParquet。
• MySQL 调优： 设置合理的 fetchSize（建议 1000-5000）；使用 rewriteBatchedStatements=true 提升写入性能；避免在高峰期读取业务主库，优先使用只读从库。
• Kafka 调优： 批量读取时设置 maxOffsetsPerTrigger 控制单批次数据量；使用 Avro 或 Protobuf 序列化替代 JSON 以减少解析开销；开启 kafka.enable.auto.commit=false 由 Spark 管理 offset。
• 通用调优： 使用 spark.sql.files.maxPartitionBytes 控制单分区大小（默认 128MB）；开启 spark.sql.adaptive.coalescePartitions.enabled 动态合并小分区。