4. 事件标准化：事件Schema设计、字段映射与清洗、事件ID生成策略

好，咱们进入第四章。这一章我打算聊聊事件标准化。

你想想看，风控系统每天要处理几千万甚至上亿条事件。这些事件来自不同渠道——App、网页、后端API、第三方数据源。格式五花八门，字段命名随心所欲。如果不做标准化，后面建模就是一场噩梦。

我个人习惯，在项目启动的第一周就把事件Schema定死。为什么？因为后面改Schema的成本太高了。我在上一家公司就吃过这个亏——上线三个月后才发现某个关键字段没定义，结果全量数据要重跑，那叫一个酸爽。

4.1 事件Schema设计：先把骨架搭好

事件Schema，说白了就是给每个事件定一个「身份证」。它规定了事件长什么样，必须包含哪些字段，每个字段是什么类型。

我一般把事件Schema分成三层：

嗯，这里要注意：基础层字段必须非空。我在项目中遇到过，有人把event_id写成可选字段，结果下游做事件去重时直接崩了。所以我的原则是——基础层字段，一个都不能少，一个都不能空。

下面是一个我常用的Schema定义示例（用JSON Schema格式）：

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "风控事件Schema",
  "type": "object",
  "required": ["event_id", "event_type", "timestamp", "source"],
  "properties": {
    "event_id": {
      "type": "string",
      "description": "全局唯一事件ID"
    },
    "event_type": {
      "type": "string",
      "enum": ["login", "register", "transaction", "withdraw"]
    },
    "timestamp": {
      "type": "integer",
      "description": "Unix毫秒时间戳"
    },
    "source": {
      "type": "string",
      "enum": ["app", "web", "api", "third_party"]
    },
    "user_id": {
      "type": "string"
    },
    "device_id": {
      "type": "string"
    },
    "ip": {
      "type": "string",
      "pattern": "^\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}$"
    },
    "amount": {
      "type": "number",
      "minimum": 0
    },
    "ext_data": {
      "type": "object"
    }
  }
}

你看，我把event_type和source都做成了枚举类型。为什么？因为这样可以避免脏数据。我曾经见过有人把event_type写成"Login"、"login"、"LOGIN"三种格式，结果统计时数据对不上。枚举一限制，这种问题就没了。

4.2 字段映射与清洗：把脏数据洗干净

字段映射，就是把不同来源的字段名统一成Schema里定义的名字。比如A系统叫"user_name"，B系统叫"username"，C系统叫"uid"，映射后统一叫"user_id"。

清洗呢，就是处理那些不合规的数据。比如时间戳格式不对、IP地址格式错误、金额为负数等等。

我一般把清洗分成四个步骤：

1. 格式标准化：统一时间格式、数字格式、字符串编码
2. 空值处理：该填默认值的填默认值，该拒绝的拒绝
3. 异常值过滤：比如IP地址不在合法范围内，直接丢弃
4. 业务规则校验：比如转账金额不能超过账户余额

举个例子，我曾经处理过一个第三方数据源，它传过来的时间戳是"2024-01-15 14:30:00"这种字符串格式。而我们的Schema要求是Unix毫秒时间戳。怎么办？写个映射函数：

def map_timestamp(raw_value, source_format="%Y-%m-%d %H:%M:%S"):
    """
    将字符串时间戳映射为Unix毫秒时间戳
    """
    try:
        dt = datetime.strptime(raw_value, source_format)
        return int(dt.timestamp() * 1000)
    except (ValueError, TypeError) as e:
        # 记录异常，返回None
        logger.warning(f"时间戳转换失败: {raw_value}, 错误: {e}")
        return None

你看，这里我加了一个异常捕获。为什么？因为数据源不可靠。我在项目中遇到过，某个数据源偶尔会传一个空字符串过来，如果不加try-catch，整个清洗流程就挂了。

4.3 事件ID生成策略：别让ID重复

事件ID，是每个事件的唯一标识。它有多重要？这么说吧，事件去重、数据关联、问题排查，全都靠它。

我见过有人直接用数据库自增ID当事件ID。嗯，这在单机环境下没问题。但分布式环境下呢？多个节点同时生成ID，冲突概率极高。

目前业界主流的ID生成策略有这么几种：

我个人习惯用Snowflake算法。为什么？因为它生成的ID是64位长整型，在数据库里做索引效率很高。而且它包含了时间戳信息，你可以直接从ID里解析出事件发生的时间，不需要额外查表。

下面是一个简化版的Snowflake实现：

class SnowflakeIdGenerator:
    def __init__(self, worker_id, datacenter_id):
        self.worker_id = worker_id
        self.datacenter_id = datacenter_id
        self.sequence = 0
        self.last_timestamp = -1
        
        # 各部分的位数
        self.worker_id_bits = 5
        self.datacenter_id_bits = 5
        self.sequence_bits = 12
        
        # 最大值
        self.max_worker_id = -1 ^ (-1 << self.worker_id_bits)
        self.max_datacenter_id = -1 ^ (-1 << self.datacenter_id_bits)
        
        # 位移量
        self.worker_id_shift = self.sequence_bits
        self.datacenter_id_shift = self.sequence_bits + self.worker_id_bits
        self.timestamp_shift = self.sequence_bits + self.worker_id_bits + self.datacenter_id_bits
        
    def next_id(self):
        timestamp = int(time.time() * 1000)
        
        if timestamp < self.last_timestamp:
            # 时钟回拨，抛出异常
            raise Exception("时钟回拨，拒绝生成ID")
        
        if timestamp == self.last_timestamp:
            self.sequence = (self.sequence + 1) & (-1 ^ (-1 << self.sequence_bits))
            if self.sequence == 0:
                # 当前毫秒序列号用完，等待下一毫秒
                while timestamp <= self.last_timestamp:
                    timestamp = int(time.time() * 1000)
        else:
            self.sequence = 0
        
        self.last_timestamp = timestamp
        
        return ((timestamp - 1609459200000) << self.timestamp_shift) | \
               (self.datacenter_id << self.datacenter_id_shift) | \
               (self.worker_id << self.worker_id_shift) | \
               self.sequence

你看，这里我处理了时钟回拨的问题。为什么？因为Snowflake算法依赖系统时钟，如果时钟回拨，生成的ID可能会重复。我在生产环境中就遇到过——某台服务器的NTP同步导致时钟回拨了1秒，结果生成了几百个重复ID。从那以后，我就在代码里加了时钟回拨检测。

4.4 本章知识体系

说了这么多，我画了一张图来总结本章的核心逻辑。你看，事件标准化其实就三件事：定Schema、做映射、生成ID。这三件事做好了，后面的风控模型才能跑得稳。

这张图把整个流程串起来了。你从左边看，原始事件进来，经过三步处理，变成标准化事件出去。每一步都有对应的技术选型和注意事项。

好了，这一章就到这里。记住：事件标准化不是一次性工作，而是一个持续迭代的过程。随着业务发展，你会不断遇到新的数据源、新的字段、新的清洗规则。但只要Schema的骨架搭好了，后面加东西就简单多了。