📚 课程目录第4章数据采集与接入

4. 数据采集与接入：实时数据源

实时风险拦截的核心在于“快”——从风险事件发生到拦截决策完成，端到端延迟通常要求控制在 200ms 以内。这要求数据采集层具备高吞吐、低延迟、可回放的能力。本章聚焦生产环境最主流的两个实时数据源组件：Kafka 与 Flume，并给出 Python 连接 Kafka 的完整示例及数据格式规范。

4.1 实时数据源选型对比

特性	Apache Kafka	Apache Flume
定位	分布式消息流平台	日志采集与聚合工具
数据持久化	磁盘持久化，可配置保留策略	基于 Channel 的内存/文件缓冲
吞吐量	百万级 msg/s（集群）	万级 msg/s（单机）
数据回溯	支持 offset 重置，可重放历史数据	不支持（消费即删除）
适用场景	核心业务事件流、风险事件总线	日志采集、边缘节点数据汇聚
与 Python 集成	成熟（confluent-kafka / kafka-python）	需通过 Avro/Thrift 或自定义 Sink

选型建议：保险风控场景中，Kafka 作为事件总线是标准选择；Flume 通常部署在日志产生端（如 Nginx、应用服务器）作为前置采集层，再将数据汇入 Kafka。

4.2 实时数据流架构

下图展示保险风险实时拦截中数据采集与接入的典型拓扑：

  关键路径：应用服务器/移动端 → Flume（可选） → Kafka → 风控引擎。Kafka 作为中心缓冲层，解耦数据生产与消费，支持多个消费者以不同速率处理同一份数据。

4.3 Python 连接 Kafka 示例

生产环境推荐使用 confluent-kafka 库（基于 librdkafka，性能优于 kafka-python）。以下展示生产者与消费者的核心代码。

4.3.1 安装依赖

pip install confluent-kafka avro-python3  # avro 用于序列化

4.3.2 生产者：发送风险事件

from confluent_kafka import Producer
import json

# 配置参数
conf = {
    'bootstrap.servers': 'kafka1:9092,kafka2:9092',
    'acks': 'all',                 # 所有副本确认
    'retries': 3,
    'compression.type': 'snappy',  # 压缩减少网络开销
}

producer = Producer(conf)

def delivery_report(err, msg):
    if err is not None:
        print(f'发送失败: {err}')
    else:
        print(f'发送成功: {msg.topic()} [{msg.partition()}] @ {msg.offset()}')

# 构造风险事件
event = {
    "event_id": "evt_20250321103245_001",
    "policy_no": "POL20250321001",
    "risk_type": "fraud_claim",
    "amount": 150000.00,
    "timestamp": "2025-03-21T10:32:45+08:00",
    "source": "mobile_app"
}

producer.produce(
    topic='risk-events',
    key=event['policy_no'],
    value=json.dumps(event).encode('utf-8'),
    callback=delivery_report
)
producer.flush()  # 等待发送完成

4.3.3 消费者：实时消费风险事件

from confluent_kafka import Consumer, KafkaError

conf = {
    'bootstrap.servers': 'kafka1:9092,kafka2:9092',
    'group.id': 'risk-engine-group',
    'auto.offset.reset': 'earliest',  # 从头开始消费（测试环境）
    'enable.auto.commit': True,
    'auto.commit.interval.ms': 5000,
}

consumer = Consumer(conf)
consumer.subscribe(['risk-events'])

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            else:
                print(f'消费错误: {msg.error()}')
                break
        
        # 反序列化
        event = json.loads(msg.value().decode('utf-8'))
        print(f"收到事件: {event['event_id']} | 风险类型: {event['risk_type']}")
        
        # 此处调用风控规则引擎进行实时拦截判断
        # risk_result = risk_engine.evaluate(event)
        
finally:
    consumer.close()

生产注意事项：

消费者 group.id 必须唯一，避免与其它服务冲突。
auto.offset.reset 在生产环境建议设为 latest，避免重启时重复消费大量历史数据。
建议开启 enable.auto.commit=False，在业务处理成功后再手动提交 offset，确保“至少一次”语义。

4.4 数据格式规范

统一的数据格式是风控规则正确执行的基石。保险风险事件采用 JSON + Avro Schema 双重规范：传输层使用 JSON 便于调试，持久化层使用 Avro 保证强类型与向后兼容。

4.4.1 风险事件通用字段

字段名	类型	必填	说明
event_id	string	是	全局唯一事件 ID，格式：{来源}_{时间戳}_{序列号}
policy_no	string	是	保单号，用作 Kafka 消息 key
risk_type	string	是	风险类型枚举：fraud_claim / abnormal_application / payment_risk
amount	double	否	涉及金额（元），无金额时填 0.0
timestamp	string	是	ISO 8601 格式，含时区
source	string	是	数据来源：web_app / mobile_app / api_gateway / batch
ext_info	object	否	扩展字段，存放业务自定义属性

4.4.2 Avro Schema 示例

{
  "namespace": "com.insurance.risk.event",
  "type": "record",
  "name": "RiskEvent",
  "fields": [
    {"name": "event_id",   "type": "string"},
    {"name": "policy_no",  "type": "string"},
    {"name": "risk_type",  "type": {"type": "enum", "name": "RiskType",
                                    "symbols": ["fraud_claim", "abnormal_application", "payment_risk"]}},
    {"name": "amount",     "type": "double", "default": 0.0},
    {"name": "timestamp",  "type": "string"},
    {"name": "source",     "type": "string"},
    {"name": "ext_info",   "type": ["null", {"type": "map", "values": "string"}], "default": null}
  ]
}

规范要点：

所有时间字段统一使用 UTC+8 的 ISO 8601 字符串，避免时区歧义。
金额字段使用 double，保留两位小数；风控规则内部统一以“分”为单位计算时需自行转换。
ext_info 使用 map 类型，支持未来扩展字段而不破坏 schema 兼容性。

4.5 数据接入流程总结

完成数据采集与接入后，下一环节将进入实时风控规则引擎的构建——如何利用 Python 高效消费 Kafka 事件流并执行百毫秒级的风险决策。