4. 数据预处理：实时数据流接入（Kafka）、数据清洗、字段标准化

在实时风控场景中，数据预处理是连接原始交易事件与风控引擎的桥梁。本章聚焦于如何将海量、无序、异构的实时交易数据，通过标准化的管道转化为可供规则引擎与模型消费的高质量特征。核心环节包括：Kafka 实时流接入、数据清洗与字段标准化。

4.1 实时数据流接入（Kafka）

Kafka 作为高吞吐、低延迟的分布式消息队列，是实时风控系统的首选数据总线。交易系统将每笔交易事件（如支付、转账、登录）以消息形式发布到 Kafka Topic，风控系统作为消费者实时拉取处理。

以下为 Kafka 消费者接入实时交易流的典型代码片段（Java 风格）：

Properties props = new Properties();
props.put("bootstrap.servers", "kafka-cluster:9092");
props.put("group.id", "risk_consumer_group_v1");
props.put("enable.auto.commit", "false");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("transaction_topic", "login_topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 1. 解析 JSON 消息
        // 2. 数据清洗与标准化
        // 3. 送入风控引擎
        processTransaction(record.value());
        // 4. 业务处理成功后手动提交 offset
        consumer.commitSync();
    }
}

4.2 数据清洗

原始交易数据常包含脏数据、缺失值、异常格式等噪声，必须经过清洗才能进入后续特征计算。清洗策略需兼顾实时性（毫秒级处理）与准确性。

4.2.1 常见脏数据类型与处理策略

4.2.2 清洗流程伪代码

def clean_transaction(raw_json):
    # 1. 基础格式校验
    if not is_valid_json(raw_json):
        log_warning("Invalid JSON format")
        return None

    tx = json.loads(raw_json)

    # 2. 必填字段检查
    required_fields = ["tx_id", "user_id", "amount", "timestamp", "merchant_id"]
    for field in required_fields:
        if field not in tx or tx[field] is None:
            log_warning(f"Missing required field: {field}")
            return None

    # 3. 字段类型与范围校验
    if not isinstance(tx["amount"], (int, float)) or tx["amount"] <= 0:
        log_warning(f"Invalid amount: {tx['amount']}")
        return None

    if not is_valid_timestamp(tx["timestamp"]):
        log_warning(f"Invalid timestamp: {tx['timestamp']}")
        return None

    # 4. 幂等去重（基于 tx_id）
    if redis.exists(f"tx_idempotent:{tx['tx_id']}"):
        log_info(f"Duplicate transaction: {tx['tx_id']}")
        return None
    redis.setex(f"tx_idempotent:{tx['tx_id']}", 3600, "1")

    # 5. 返回清洗后的结构化数据
    return tx

4.3 字段标准化

不同业务系统对同一概念的字段命名、格式、单位可能不同。标准化是将异构数据统一为风控引擎可识别的内部格式，是特征计算的前提。

4.3.1 标准化映射规则

4.3.2 标准化处理示例

def standardize_fields(tx):
    std_tx = {}

    # 交易 ID 标准化
    raw_tx_id = tx.get("trans_id") or tx.get("order_id") or tx.get("payment_id")
    std_tx["tx_id"] = str(raw_tx_id).strip()

    # 金额标准化：元 → 分
    raw_amount = tx.get("amount") or tx.get("total_amount") or tx.get("money")
    std_tx["amount"] = int(round(float(raw_amount) * 100))

    # 时间标准化：统一为 Unix 毫秒
    raw_ts = tx.get("trans_time") or tx.get("create_time") or tx.get("event_time")
    std_tx["event_ts"] = parse_to_unix_ms(raw_ts)

    # IP 标准化
    raw_ip = tx.get("ip") or tx.get("client_ip") or tx.get("remote_addr")
    std_tx["client_ip"] = str(raw_ip).strip() if is_valid_ip(raw_ip) else "0.0.0.0"

    # 设备 ID 标准化
    raw_device = tx.get("device_id") or tx.get("device_fingerprint") or tx.get("imei")
    std_tx["device_id"] = str(raw_device).lower().replace(" ", "").replace("-", "")

    return std_tx