第4章数据源接入

数据源接入：Kafka Producer API实战

做实时风控系统，第一关就是数据源接入。

数据进不来，后面再牛的模型也是白搭。我这些年踩过的坑，有一半都跟数据接入有关。今天咱们就聊聊怎么用Kafka Producer把数据稳稳地送进去。

为什么选Kafka？

说白了，风控场景对消息队列的要求就三点：高吞吐、低延迟、不丢数据。Kafka在这块儿做得确实不错。我曾在某支付公司做过一个项目，每秒要处理上万笔交易，Kafka扛得住。

你想想看，如果交易数据丢了，那可不是小事。Kafka的持久化机制能保证数据不丢，这点很关键。

模拟交易数据生成器

实战之前，得先有数据。我习惯写一个模拟生成器，这样调试起来方便。

import json
import time
import random
from datetime import datetime

class TransactionGenerator:
    """模拟交易数据生成器"""
    
    def __init__(self):
        self.user_ids = [f'U{str(i).zfill(6)}' for i in range(1, 10001)]
        self.merchant_ids = [f'M{str(i).zfill(4)}' for i in range(1, 501)]
        self.device_types = ['iOS', 'Android', 'Web', 'H5']
        self.channels = ['alipay', 'wechat', 'card', 'balance']
        
    def generate(self):
        """生成一条交易记录"""
        return {
            'event_id': f'E{int(time.time()*1000)}{random.randint(1000,9999)}',
            'user_id': random.choice(self.user_ids),
            'merchant_id': random.choice(self.merchant_ids),
            'amount': round(random.uniform(0.01, 50000.00), 2),
            'device_type': random.choice(self.device_types),
            'channel': random.choice(self.channels),
            'ip': f'{random.randint(1,255)}.{random.randint(0,255)}.{random.randint(0,255)}.{random.randint(1,255)}',
            'timestamp': datetime.now().isoformat(),
            'risk_score': round(random.uniform(0, 100), 2)
        }

# 用起来很简单
gen = TransactionGenerator()
for _ in range(5):
    print(json.dumps(gen.generate(), indent=2))
    time.sleep(0.5)

小提示：生成器里的user_id和merchant_id最好用真实分布。我在项目中遇到过，如果数据分布太均匀，后面测试规则时会发现不了问题。

JSON vs AVRO：怎么选？

这个问题我经常被问到。直接说结论：

特性	JSON	AVRO
可读性	好，肉眼能看懂	差，二进制格式
序列化速度	较慢	快，约快3-5倍
数据体积	大，有冗余字段名	小，压缩率高
Schema演进	灵活，字段可增可减	需配合Schema Registry
适用场景	调试、小规模数据	生产环境、大规模数据

我个人习惯是：开发调试用JSON，上线后切到AVRO。为什么呢？因为AVRO的Schema Registry能保证生产者和消费者之间的数据契约。我曾经吃过亏，上游改了字段类型，下游直接崩了。用了Schema Registry之后，这种问题再没出现过。

AVRO序列化实战

先定义Schema文件。这是交易数据的AVRO schema：

{
  "namespace": "com.risk.trade",
  "type": "record",
  "name": "TradeEvent",
  "fields": [
    {"name": "event_id", "type": "string"},
    {"name": "user_id", "type": "string"},
    {"name": "merchant_id", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "device_type", "type": "string"},
    {"name": "channel", "type": "string"},
    {"name": "ip", "type": "string"},
    {"name": "timestamp", "type": "string"},
    {"name": "risk_score", "type": "double", "default": 0.0}
  ]
}

然后写Producer代码：

from confluent_kafka import avro, SerializingProducer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer

# 配置Schema Registry
schema_registry_conf = {'url': 'http://localhost:8081'}
schema_registry_client = SchemaRegistryClient(schema_registry_conf)

# 加载Schema
with open('trade_event.avsc', 'r') as f:
    schema_str = f.read()

avro_serializer = AvroSerializer(
    schema_registry_client,
    schema_str,
    to_dict=lambda obj, ctx: obj
)

# 配置Producer
producer_conf = {
    'bootstrap.servers': 'localhost:9092',
    'key.serializer': avro_serializer,
    'value.serializer': avro_serializer
}

producer = SerializingProducer(producer_conf)

# 发送数据
def send_trade_event(event_data):
    try:
        producer.produce(
            topic='trade-events',
            key={'event_id': event_data['event_id']},
            value=event_data,
            on_delivery=lambda err, msg: 
                print(f'发送成功: {msg.key()}') if not err 
                else print(f'发送失败: {err}')
        )
        producer.poll(0)  # 触发回调
    except Exception as e:
        print(f'序列化异常: {e}')

# 批量发送
gen = TransactionGenerator()
for i in range(100):
    event = gen.generate()
    send_trade_event(event)
    if i % 10 == 0:
        producer.flush()  # 每10条flush一次

producer.flush()

注意：AVRO的Schema兼容性有讲究。我建议用FORWARD兼容模式，这样新增字段不会影响老消费者。曾经有个团队用了BACKWARD模式，结果升级Schema后老数据全读不了，那叫一个惨。

Schema Registry集成

Schema Registry不是可选项，是必需品。为什么？

版本管理：每个Schema都有版本号，改了能追溯
兼容性检查：发布新Schema前自动校验，防止破坏性变更
数据契约：生产者和消费者用同一份Schema，不会出现鸡同鸭讲

集成步骤其实不复杂：

部署Schema Registry服务（一般用Confluent的）
在Producer和Consumer中配置Schema Registry地址
定义好AVRO Schema文件
用AvroSerializer/Deserializer处理数据

嗯，这里要注意一点：Schema Registry的URL不要写死在代码里。我习惯用环境变量或者配置中心来管理，这样不同环境（开发、测试、生产）可以灵活切换。

知识体系总览

下面这张图能帮你理清思路：

这张图把整个数据接入流程串起来了。从数据生成到序列化，再到Schema Registry管理，最后通过Producer发到Kafka。每一步都有它的职责，缺一不可。

核心要点：

生产环境一定要用AVRO + Schema Registry
模拟数据生成器要贴近真实分布
Producer要处理好异常和重试逻辑
Schema兼容性选FORWARD模式最稳妥

好了，数据源接入这块就聊到这儿。记住一句话：数据进得稳，风控才靠谱。下一节咱们聊聊怎么从Kafka里把数据消费出来做实时计算。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

数据源接入：Kafka Producer API实战

为什么选Kafka？

模拟交易数据生成器

JSON vs AVRO：怎么选？

AVRO序列化实战

Schema Registry集成

知识体系总览