第四章：数据接入层——多源异构数据接入、实时流处理与批处理

数据接入层，说白了就是风控系统的“嘴巴”。

你想想看，一个图谱风控系统，要喂进去的数据五花八门——交易流水、用户行为、设备指纹、黑名单、工商信息……这些数据格式不同、时效要求不同、量级也不同。怎么把它们统一接进来，还要保证不丢、不错、不延迟？这就是本章要聊的事。

4.1 多源异构数据接入：统一入口的设计思路

我刚开始做风控系统时，犯过一个低级错误——每种数据源单独写一个接入模块。结果呢？维护成本爆炸，改一个接口要动七八个地方。

后来我学乖了：统一接入层，所有数据源走同一个管道。

具体怎么做？我习惯用三层抽象：

• 数据源适配器：每种数据源（MySQL、Kafka、API、文件）写一个适配器，负责协议转换
• 统一消息模型：所有数据转成标准JSON格式，包含元数据字段（source、timestamp、event_type）
• 路由分发层：根据event_type，把数据路由到实时流或批处理通道

举个例子，一个交易事件接入后的标准格式：

{
  "source": "trade_system",
  "timestamp": 1712345678000,
  "event_type": "transaction",
  "payload": {
    "trade_id": "T20240315001",
    "user_id": "U10086",
    "amount": 5000.00,
    "merchant_id": "M2022"
  }
}

嗯，这里要注意：元数据字段不能少。我在项目中遇到过，某个数据源忘了带时间戳，结果回溯分析时根本对不上时间线，排查了整整两天。

4.2 实时流处理：Kafka 的实战配置

实时流处理，Kafka 是绕不开的。说白了，它就是风控系统的“高速公路”。

我个人习惯用 Kafka 做两件事：

• 事件总线：所有实时数据先进 Kafka，下游消费方各取所需
• 削峰填谷：风控系统扛不住瞬时流量，Kafka 做缓冲

4.2.1 主题设计

我建议按业务域划分主题，别搞一个大而全的 topic。比如：

主题名	用途	分区数
risk-trade-events	交易事件	12
risk-user-behavior	用户行为	8
risk-device-fingerprint	设备指纹	6
risk-blacklist-update	黑名单变更	3

分区数怎么定？我有个经验公式：分区数 = 目标吞吐量(MB/s) / 单分区吞吐量(MB/s)。单分区吞吐量一般按 10MB/s 估算，留 30% 余量。

4.2.2 消费者配置避坑

关键配置项：

• max.poll.interval.ms：建议 300000（5分钟），别低于 180000
• max.poll.records：建议 500，根据单条数据大小调整
• enable.auto.commit：建议 false，手动提交更可控

手动提交的代码片段：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        // 处理业务逻辑
        processEvent(record.value());
    }
    // 处理完一批再提交
    consumer.commitSync();
}

4.3 批处理：Spark 的离线清洗

实时流处理的是“热数据”，批处理管的是“冷数据”。

我一般在凌晨用 Spark 做三件事：

• 历史数据回刷：修复之前接入的错误数据
• 批量特征计算：比如用户 30 天交易频次、金额统计
• 数据质量校验：检查缺失率、异常值、重复数据

4.3.1 Spark 作业模板

这是我常用的批处理框架：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, count

spark = SparkSession.builder \
    .appName("RiskDataClean") \
    .config("spark.sql.shuffle.partitions", "200") \
    .getOrCreate()

# 读取原始数据
df = spark.read.parquet("/data/raw/trade/20240315/")

# 数据清洗
cleaned_df = df \
    .dropDuplicates(["trade_id"]) \
    .filter(col("amount") > 0) \
    .fillna({"user_id": "unknown", "merchant_id": "unknown"})

# 质量报告
quality_report = cleaned_df.agg(
    count("*").alias("total_records"),
    (count(when(col("user_id") == "unknown", 1)) / count("*") * 100).alias("missing_user_pct")
)

quality_report.show()
cleaned_df.write.mode("overwrite").parquet("/data/clean/trade/20240315/")

4.4 数据清洗策略：脏数据的“三刀流”

数据清洗，说白了就是“去脏、补缺、纠错”。

我总结了一套“三刀流”策略：

1. 第一刀：去重
   • 精确去重：基于主键（trade_id、event_id）
   • 模糊去重：基于时间窗口+相似度（比如 1 秒内同一用户同一金额）
2. 第二刀：补缺
   • 必填字段缺失：直接丢弃或标记为“待补录”
   • 可选字段缺失：用默认值填充（如金额缺失填 0）
   • 时间字段缺失：用 Kafka 消息时间戳替代
3. 第三刀：纠错
   • 格式校验：手机号 11 位、身份证 18 位
   • 逻辑校验：交易时间不能晚于当前时间
   • 范围校验：金额不能为负数，年龄不能超过 150

4.5 整体架构图

下面这张图，是我做过的风控系统数据接入层的标准架构。你一看就明白了：

这张图里，数据从左边进来，经过统一接入层，分流到实时流和批处理两条线，最终汇入图数据库。实时流处理秒级数据，批处理处理 T+1 数据，互不干扰。

4.6 总结一下

数据接入层，说白了就是“接得住、洗得净、分得清”。

• 接得住：Kafka 扛住峰值流量
• 洗得净：三刀流策略去脏补缺
• 分得清：实时流和批处理各司其职

我做了这么多年风控，发现 80% 的问题都出在数据接入层。数据没接好，后面的图计算、规则引擎全是白搭。所以，这一层值得你花时间打磨。