📚 课程目录第3章风控数据源接入

3. 风控数据源接入：从HDFS读取风控日志、从Kafka读取实时风控事件、从JDBC读取用户画像数据

风控实验的数据源通常呈现“批量 + 实时 + 维度”的三元结构。本章聚焦三种核心接入方式：HDFS 承载历史风控日志的批量读取，Kafka 提供实时风控事件的流式消费，JDBC 用于查询用户画像等维度表。三种方式在 Spark 中分别对应 spark.read、spark.readStream 和 spark.read.format("jdbc")。

实验前提：确保 Spark 集群已正确配置 HDFS 客户端、Kafka 客户端库（如 spark-sql-kafka-0-10）以及对应数据库的 JDBC 驱动（如 mysql-connector-java）。

3.1 从 HDFS 读取风控日志

风控日志通常以 Parquet、ORC 或压缩文本格式存储在 HDFS 上。Spark 通过 DataSource API 实现高效批量读取，支持分区裁剪和谓词下推。

3.1.1 典型读取方式

存储格式	读取方式	适用场景
Parquet	`spark.read.parquet("hdfs://...")`	列式存储，压缩率高，风控日志首选
ORC	`spark.read.orc("hdfs://...")`	Hive 生态兼容，支持 ACID
JSON / CSV	`spark.read.json("hdfs://...")`	调试或小规模数据
Text	`spark.read.text("hdfs://...")`	原始日志行解析

3.1.2 代码示例：读取 Parquet 格式风控日志

// 读取 HDFS 上的风控日志（Parquet 格式）
val riskLogDF = spark.read
  .option("basePath", "/data/risk_logs")
  .parquet("/data/risk_logs/dt=2025-03-01/*.parquet")

riskLogDF.printSchema()
riskLogDF.createOrReplaceTempView("risk_logs")

  性能要点：使用 basePath 配合分区目录（如 dt=）可让 Spark 自动推断分区列，避免全量扫描。

3.2 从 Kafka 读取实时风控事件

实时风控场景下，Kafka 作为事件总线承载交易、登录、设备指纹等流式数据。Spark Structured Streaming 通过 readStream 接口以微批或连续模式消费 Kafka 消息。

3.2.1 核心配置参数

参数	说明	推荐值
`kafka.bootstrap.servers`	Kafka 集群地址	`node1:9092,node2:9092`
`subscribe`	订阅的 topic 列表	`risk_event`
`startingOffsets`	起始偏移量策略	`earliest` / `latest`
`maxOffsetsPerTrigger`	每批次最大消息数	`10000`
`failOnDataLoss`	数据丢失时是否失败	`false`（生产环境建议 `true`）

3.2.2 代码示例：消费实时风控事件

val kafkaStreamDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "node1:9092,node2:9092")
  .option("subscribe", "risk_event")
  .option("startingOffsets", "earliest")
  .option("maxOffsetsPerTrigger", 10000)
  .option("failOnDataLoss", "false")
  .load()

// 解析 Kafka 消息的 key/value（二进制格式需转字符串）
val parsedStreamDF = kafkaStreamDF
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "topic", "partition", "offset", "timestamp")

parsedStreamDF.printSchema()

注意：Kafka 消息的 key 和 value 默认是二进制类型，必须显式转换为 STRING 或使用 from_json 解析为结构化字段。

3.3 从 JDBC 读取用户画像数据

用户画像数据（如年龄、职业、历史逾期次数）通常存储在 MySQL、PostgreSQL 或 Oracle 中。Spark 通过 JDBC 连接器以分区并行方式读取，避免单点压力。

3.3.1 分区读取策略

参数	作用	示例
`partitionColumn`	分区列（通常为数字或日期）	`user_id`
`lowerBound`	分区下界	`1`
`upperBound`	分区上界	`10000000`
`numPartitions`	并行分区数	`10`
`fetchSize`	每次拉取行数	`10000`

3.3.2 代码示例：读取用户画像表

val userProfileDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql://db_host:3306/risk_db?useSSL=false")
  .option("dbtable", "user_profile")
  .option("user", "risk_app")
  .option("password", "******")
  .option("driver", "com.mysql.cj.jdbc.Driver")
  .option("partitionColumn", "user_id")
  .option("lowerBound", 1)
  .option("upperBound", 10000000)
  .option("numPartitions", 10)
  .option("fetchSize", 10000)
  .load()

userProfileDF.createOrReplaceTempView("user_profile")

  最佳实践：务必设置 partitionColumn 和 numPartitions，否则 Spark 会使用单线程读取，导致 OOM 或超时。

3.4 三种数据源接入架构总览

下图展示了风控实验数据接入的整体架构：HDFS 提供批量历史日志，Kafka 提供实时事件流，JDBC 提供维度画像，三者最终汇入 Spark 计算层进行特征加工与模型推理。

3.5 三种接入方式对比总结

维度	HDFS 风控日志	Kafka 实时事件	JDBC 用户画像
数据性质	批量历史数据	实时流式数据	静态维度数据
读取方式	`spark.read`	`spark.readStream`	`spark.read.format("jdbc")`
典型格式	Parquet / ORC	JSON / Avro	关系表行
分区策略	目录分区 + 文件拆分	Kafka 分区 + 微批	JDBC 分区列 + numPartitions
容错机制	文件级重试	偏移量管理 + checkpoint	连接池 + 重试
适用阶段	离线特征 / 回溯分析	实时决策 / 事件驱动	画像补充 / 维度扩展

实验建议：在同一个 SparkSession 中，可以同时注册 HDFS 表、Kafka 流表和 JDBC 表，然后通过 SQL 或 DataFrame API 进行关联分析，实现“批流一体”的风控特征计算。