4. Flink DataStream API 进阶：Source、Transformation 与 Sink

在实时风控规则引擎中，Source（数据接入）、Transformation（计算逻辑）与 Sink（结果输出）构成了 DataStream API 的核心三要素。本章将深入讲解这三类算子的高级用法，并结合风控场景给出最佳实践。

4.1 Source：数据接入层

Source 是 Flink 流处理的起点，负责从外部系统读取数据并转换为 DataStream。在风控场景中，最常见的 Source 是 Kafka，同时也需要支持自定义 Source 以适配非标准协议。

4.1.1 Kafka Source（推荐方式）

Flink 1.13+ 推荐使用 KafkaSource（新 API），相比旧的 FlinkKafkaConsumer 具有更好的容错性和水位线支持。

配置项	说明	风控推荐值
bootstrap.servers	Kafka 集群地址	根据环境配置
topics	订阅的 Topic 列表	风控事件 Topic
group.id	消费者组 ID	risk-engine-group
offsets.commit	偏移量提交方式	手动提交（配合 Checkpoint）
value.deserializer	反序列化器	JSON 或 Avro 反序列化

// KafkaSource 示例（风控事件接入）
KafkaSource<RiskEvent> source = KafkaSource.<RiskEvent>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("risk-events")
    .setGroupId("risk-engine")
    .setStartingOffsets(OffsetsInitializer.latest())
    .setValueOnlyDeserializer(new RiskEventDeserializer())
    .build();

DataStream<RiskEvent> eventStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

4.1.2 自定义 Source

当需要接入非标准协议（如 MQTT、WebSocket、UDP）或模拟测试数据时，可通过实现 SourceFunction（旧 API）或 RichSourceFunction 来构建自定义 Source。

// 自定义 Source：模拟风控事件流
public class MockRiskEventSource extends RichSourceFunction<RiskEvent> {
    private volatile boolean running = true;

    @Override
    public void run(SourceContext<RiskEvent> ctx) throws Exception {
        Random random = new Random();
        while (running) {
            RiskEvent event = new RiskEvent(
                "user_" + random.nextInt(1000),
                random.nextDouble() > 0.8 ? "login" : "payment",
                System.currentTimeMillis()
            );
            ctx.collect(event);
            Thread.sleep(100); // 模拟 100ms 间隔
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}

4.2 Transformation：核心计算算子

Transformation 是规则引擎的“计算心脏”，负责对数据流进行过滤、转换、分组等操作。以下四个算子是风控场景中使用频率最高的。

4.2.1 map：一对一转换

map 将输入元素转换为另一个元素，常用于数据清洗、字段映射、类型转换。

// 将原始事件转换为规则引擎内部格式
DataStream<InternalEvent> internalStream = eventStream
    .map(new MapFunction<RiskEvent, InternalEvent>() {
        @Override
        public InternalEvent map(RiskEvent event) {
            return new InternalEvent(event.getUserId(), event.getEventType(), event.getTimestamp());
        }
    });

4.2.2 filter：条件过滤

filter 根据条件保留或丢弃元素，在风控中用于白名单过滤、事件类型筛选、降噪处理。

// 只保留支付类事件，过滤掉非关键事件
DataStream<RiskEvent> paymentStream = eventStream
    .filter(event -> "payment".equals(event.getEventType()));

4.2.3 flatMap：一对多转换

flatMap 可将一个输入元素转换为 0 个、1 个或多个输出元素，适用于事件拆分、规则匹配结果展开。

// 将复合事件拆分为多个原子事件
DataStream<AtomicEvent> atomicStream = eventStream
    .flatMap(new FlatMapFunction<RiskEvent, AtomicEvent>() {
        @Override
        public void flatMap(RiskEvent event, Collector<AtomicEvent> out) {
            if (event.getSubEvents() != null) {
                for (SubEvent sub : event.getSubEvents()) {
                    out.collect(new AtomicEvent(event.getUserId(), sub.getType(), sub.getValue()));
                }
            }
        }
    });

4.2.4 keyBy：分组分流

keyBy 根据指定 Key 对数据流进行逻辑分区，是后续窗口聚合、状态管理的基础。在风控中，通常按 userId、deviceId 或 ip 进行分组。

// 按用户 ID 分组，实现用户级规则隔离
KeyedStream<RiskEvent, String> keyedStream = eventStream
    .keyBy(RiskEvent::getUserId);

4.3 Sink：结果输出层

Sink 负责将处理结果写入外部系统。风控引擎的输出通常包括：告警记录、决策结果、统计指标等。

4.3.1 Kafka Sink

将结果写回 Kafka 是风控架构中的常见模式，用于下游告警系统或数据湖消费。

// KafkaSink 示例（输出决策结果）
KafkaSink<RiskResult> sink = KafkaSink.<RiskResult>builder()
    .setBootstrapServers("kafka:9092")
    .setRecordSerializer(new RiskResultSerializer("risk-results"))
    .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
    .build();

resultStream.sinkTo(sink);

4.3.2 MySQL Sink（JDBC Connector）

使用 Flink JDBC Connector 将结果写入 MySQL，适用于规则命中记录持久化。

// JDBC Sink 示例
JdbcExecutionOptions execOptions = JdbcExecutionOptions.builder()
    .withBatchSize(100)
    .withBatchIntervalMs(200)
    .build();

JdbcSink.sink(
    "INSERT INTO risk_hit_log(user_id, rule_id, hit_time) VALUES (?, ?, ?)",
    (statement, result) -> {
        statement.setString(1, result.getUserId());
        statement.setString(2, result.getRuleId());
        statement.setLong(3, result.getHitTime());
    },
    execOptions,
    new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:mysql://localhost:3306/risk_db")
        .withDriverName("com.mysql.cj.jdbc.Driver")
        .withUsername("risk_user")
        .withPassword("risk_pass")
        .build()
);

4.3.3 自定义 Sink

当需要写入 Redis、Elasticsearch 或自研存储时，可实现 SinkFunction 或继承 RichSinkFunction。

// 自定义 Sink：写入 Redis（伪代码示例）
public class RedisSink extends RichSinkFunction<RiskResult> {
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) {
        jedis = new Jedis("redis-host", 6379);
    }

    @Override
    public void invoke(RiskResult value, Context context) {
        jedis.set("risk:" + value.getUserId(), value.toJson());
    }

    @Override
    public void close() {
        if (jedis != null) jedis.close();
    }
}

4.4 完整数据流架构图

下图展示了风控规则引擎中 Source、Transformation、Sink 的完整数据流转路径：

4.5 算子选择对比表

算子	输入 → 输出	风控典型用途	注意事项
map	1 → 1	事件格式转换、字段脱敏	避免在 map 中执行耗时操作
filter	1 → 0/1	白名单过滤、事件类型筛选	条件应幂等，避免状态依赖
flatMap	1 → 0..n	复合事件拆分、规则匹配展开	注意控制输出数量，防止 OOM
keyBy	N → N	按用户/设备分组，实现规则隔离	Key 分布要均匀，避免数据倾斜

4.6 本章小结

• Source：Kafka Source 是风控引擎的标准接入方式，自定义 Source 用于扩展非标准协议。
• Transformation：map/filter/flatMap/keyBy 是规则引擎最核心的四个算子，覆盖了数据清洗、过滤、拆分、分组全流程。
• Sink：Kafka Sink 用于结果回传，MySQL Sink 用于持久化，自定义 Sink 用于适配特殊存储。
• 架构设计：Source → Transformation → Sink 的三层架构清晰解耦，便于规则引擎的扩展和维护。