📚 课程目录第4章风控数据源接入

4. 风控数据源接入：Kafka数据源配置、自定义Source、数据格式解析（JSON/CSV/Avro）

实时风控特征工程的第一步是稳定、高效地接入业务数据。本章聚焦于 Flink 与三种主流数据源的集成方式：Kafka（高吞吐消息队列）、自定义 Source（适配私有协议或特殊存储）以及多格式解析（JSON / CSV / Avro）。掌握这些接入手段，是构建可靠特征管道的基础。

4.1 Kafka 数据源配置与消费语义

Kafka 是实时风控中最常用的数据总线。Flink 通过 FlinkKafkaConsumer（旧版）或 KafkaSource（新版 API）实现 Exactly-Once 语义下的数据消费。以下为生产级配置要点：

配置项	推荐值	说明
`bootstrap.servers`	kafka1:9092,kafka2:9092	至少两个 Broker 地址，避免单点
`group.id`	risk_feature_group_v1	消费者组 ID，用于 offset 管理与重启续接
`auto.offset.reset`	earliest	新组从头消费，确保历史数据不丢失
`enable.auto.commit`	false	关闭自动提交，由 Flink 的 Checkpoint 控制 offset
`flink.partition-discovery.interval-millis`	30000	每 30s 动态发现新增分区，适应 Topic 扩容

生产建议：风控场景对延迟敏感，建议将 fetch.min.bytes 设为 1（不等待批量化），并配合 flink.checkpointing.interval（如 10s）实现低延迟 + 精确一次消费。

4.2 自定义 Source：适配私有协议与特殊存储

当数据源不在 Kafka 或标准消息队列中（如 Redis 流、RocketMQ、自研网关），需要实现 SourceFunction（旧 API）或 RichParallelSourceFunction / Source（新 API）。以下为自定义 Source 的核心结构：

public class RiskCustomSource extends RichParallelSourceFunction<String> {
    private volatile boolean running = true;
    private transient SocketChannel channel;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化连接：TCP / Redis / 文件句柄
        channel = SocketChannel.open(new InetSocketAddress("192.168.1.100", 9999));
        channel.configureBlocking(false);
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        ByteBuffer buffer = ByteBuffer.allocate(4096);
        while (running) {
            int bytesRead = channel.read(buffer);
            if (bytesRead > 0) {
                buffer.flip();
                byte[] data = new byte[buffer.remaining()];
                buffer.get(data);
                String line = new String(data, StandardCharsets.UTF_8);
                // 使用锁或同步机制保证 Checkpoint 安全
                synchronized (ctx.getCheckpointLock()) {
                    ctx.collect(line);
                }
                buffer.clear();
            } else {
                Thread.sleep(10); // 避免空转
            }
        }
    }

    @Override
    public void cancel() {
        running = false;
        try {
            if (channel != null) channel.close();
        } catch (IOException ignored) {}
    }
}

避坑提醒：自定义 Source 必须正确处理 CheckpointLock，否则在故障恢复时可能出现重复或丢失数据。建议继承 RichParallelSourceFunction 并显式使用 ctx.getCheckpointLock() 包裹 collect 操作。

4.3 数据格式解析：JSON / CSV / Avro

风控原始日志通常以 JSON、CSV 或 Avro 格式写入 Kafka。Flink 提供了对应的反序列化器，但生产环境常需自定义解析逻辑以处理脏数据、字段映射和类型转换。

4.3.1 JSON 解析

使用 JSONKeyValueDeserializationSchema 或自定义 AbstractDeserializationSchema 配合 Jackson / Fastjson：

DataStream<RiskEvent> riskStream = env
    .addSource(kafkaSource)
    .map(jsonStr -> {
        ObjectMapper mapper = new ObjectMapper();
        JsonNode root = mapper.readTree(jsonStr);
        return new RiskEvent(
            root.get("userId").asText(),
            root.get("eventType").asText(),
            root.get("amount").asDouble(),
            root.get("timestamp").asLong()
        );
    })
    .name("json_parse");

  性能优化：在 map 中复用 ObjectMapper（线程安全），避免每次 new 实例；对于嵌套 JSON，使用 treeToValue 直接转为 POJO 可减少手动字段提取。

4.3.2 CSV 解析

CSV 格式常见于传统风控系统导出的日志。Flink 内置 CsvReaderFormat，但更灵活的方式是使用 FlatMapFunction 配合正则或分割：

DataStream<RiskEvent> csvStream = kafkaSource
    .flatMap((String line, Collector<RiskEvent> out) -> {
        String[] fields = line.split(",", -1); // 保留空字段
        if (fields.length < 4) return; // 过滤异常行
        try {
            out.collect(new RiskEvent(
                fields[0].trim(),
                fields[1].trim(),
                Double.parseDouble(fields[2]),
                Long.parseLong(fields[3])
            ));
        } catch (NumberFormatException e) {
            // 记录脏数据到侧输出流
        }
    })
    .name("csv_parse");

4.3.3 Avro 解析

Avro 是风控特征工程中推荐的序列化格式，支持 Schema 演化、压缩率高。Flink 通过 ConfluentRegistryAvroDeserializationSchema 或 AvroDeserializationSchema 直接反序列化为 SpecificRecord：

// 使用 Confluent Schema Registry
DeserializationSchema<RiskEventAvro> avroSchema = 
    ConfluentRegistryAvroDeserializationSchema
        .forSpecific(RiskEventAvro.class, "http://schema-registry:8081");

DataStream<RiskEventAvro> avroStream = env
    .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "avro-source")
    .map(avroRecord -> {
        // Avro 自动生成 SpecificRecord，直接使用 get() 方法
        return new RiskEvent(
            avroRecord.getUserId().toString(),
            avroRecord.getEventType().toString(),
            avroRecord.getAmount(),
            avroRecord.getTimestamp()
        );
    });

选型建议：若团队已有 Schema Registry 基础设施，优先使用 Avro；若数据源为第三方 API 回调，JSON 更通用；CSV 仅用于遗留系统迁移，不建议在新项目中使用。

4.4 数据源接入架构总览

下图展示了从原始数据源到 Flink 内部统一事件流的完整链路，涵盖三种接入方式与三种格式解析的协同关系：

4.5 多格式混合接入策略

实际风控系统中，同一 Topic 可能包含多种格式（如旧版 CSV 与新版 JSON 并存）。推荐使用 格式嗅探 + 侧输出流 策略：

格式嗅探：在 RichMapFunction 中根据首字符（{ 为 JSON，字母/数字为 CSV，Avro 需通过 Schema Registry 判断）决定解析路径。
侧输出流：将无法解析的脏数据发送到 OutputTag，单独写入死信队列或日志，不影响主流程。
Schema 版本兼容：Avro 场景下，使用 GenericRecord 配合 SchemaRegistryClient 自动兼容字段新增/删除。

  最佳实践：在 Flink 作业启动前，通过 KafkaDeserializationSchema 的 isEndOfStream 方法返回 false，确保作业不会因空 Topic 而退出；同时设置 idleTimeout 避免无数据时 Watermark 停滞。

4.6 本章小结

本章覆盖了实时风控特征工程中数据源接入的三个核心维度：Kafka 生产级配置、自定义 Source 的 Checkpoint 安全实现、以及 JSON/CSV/Avro 三种格式的解析与选型。掌握这些内容后，你可以将任意业务数据稳定地转化为统一的 RiskEvent 流，为后续的特征计算与规则引擎提供高质量输入。