3. Flink DataStream API：Source、Transformation、Sink 核心算子实战

在实时风控巡检中，Flink DataStream API 是构建数据管道的核心工具。本章将深入解析 Source（数据源）、Transformation（转换算子）和 Sink（数据汇）三大组件，并结合风控场景演示关键算子的实战用法。

3.1 三大组件概览

DataStream API 的作业通常遵循“Source → Transformation → Sink”的流水线结构。下表总结了各组件的职责与风控场景中的典型选型：

3.2 Source：接入实时交易流

Flink 提供了丰富的内置 Source，同时也支持通过 SourceFunction 或 RichSourceFunction 实现自定义数据源。在风控场景中，最常用的是 Kafka Source 和 Socket Source（用于测试）。

3.2.1 Kafka Source 实战

以下代码演示如何从 Kafka 消费交易事件，并反序列化为 TransactionEvent POJO：

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "risk-inspector");

DataStream<String> rawStream = env.addSource(
    new FlinkKafkaConsumer<>("transaction-topic", new SimpleStringSchema(), props)
);

// 反序列化为 TransactionEvent
DataStream<TransactionEvent> eventStream = rawStream
    .map(json -> JsonUtil.parse(json, TransactionEvent.class));

3.2.2 自定义 Source：模拟交易数据

在开发与测试阶段，常使用自定义 Source 模拟高频交易流：

public class MockTransactionSource extends RichSourceFunction<TransactionEvent> {
    private volatile boolean running = true;

    @Override
    public void run(SourceContext<TransactionEvent> ctx) throws Exception {
        Random rand = new Random();
        while (running) {
            TransactionEvent event = new TransactionEvent(
                "user_" + rand.nextInt(1000),
                rand.nextDouble() * 10000,
                System.currentTimeMillis()
            );
            ctx.collect(event);
            Thread.sleep(100); // 每秒10条
        }
    }

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

3.3 Transformation：核心转换算子

Transformation 是风控逻辑的承载层。以下算子覆盖了从简单过滤到复杂状态计算的常见需求。

3.3.1 基础算子：map / filter / flatMap

• map：一对一转换，如将 JSON 字符串转为 POJO。
• filter：过滤掉低风险交易，保留金额大于阈值的记录。
• flatMap：一对多转换，如将一笔交易拆分为多条风控规则命中记录。

// 过滤出金额大于10000的高风险交易
DataStream<TransactionEvent> highRiskStream = eventStream
    .filter(event -> event.getAmount() > 10000);

// 将交易映射为告警对象
DataStream<Alert> alertStream = highRiskStream
    .map(event -> new Alert(event.getUserId(), "HIGH_AMOUNT", event.getAmount()));

3.3.2 分组与聚合：keyBy / window

风控巡检常需要按用户或设备维度进行统计。keyBy 将数据分组，window 定义时间窗口进行聚合：

// 按用户ID分组，统计每5分钟内的交易总额
DataStream<UserAggregate> aggStream = eventStream
    .keyBy(TransactionEvent::getUserId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new TransactionAggregateFunction());

3.3.3 状态算子：RichMapFunction / ValueState

在风控中，需要跨事件记录用户状态（如历史交易次数、累计金额）。Flink 的状态算子提供了容错的状态管理能力：

public class UserTransactionCountFunc extends RichMapFunction<TransactionEvent, UserStat> {
    private transient ValueState<Long> countState;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Long> descriptor =
            new ValueStateDescriptor<>("txn-count", Types.LONG);
        countState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public UserStat map(TransactionEvent event) throws Exception {
        Long currentCount = countState.value();
        if (currentCount == null) currentCount = 0L;
        currentCount++;
        countState.update(currentCount);
        return new UserStat(event.getUserId(), currentCount);
    }
}

3.4 Sink：输出告警与统计结果

Sink 负责将风控结果写入外部系统。生产环境推荐使用 Kafka Sink 或 JDBC Sink，测试阶段可使用 Print Sink 或 Socket Sink。

3.4.1 Kafka Sink 实战

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;

DataStream<Alert> alertStream = ...; // 风控告警流

FlinkKafkaProducer<Alert> kafkaSink = new FlinkKafkaProducer<>(
    "alert-topic",
    new AlertSerializationSchema(),
    props,
    FlinkKafkaProducer.Semantic.EXACTLY_ONCE
);

alertStream.addSink(kafkaSink);

3.4.2 JDBC Sink：写入风控结果表

使用 JDBCOutputFormat 或自定义 RichSinkFunction 将聚合结果写入 MySQL：

aggStream.addSink(new RichSinkFunction<UserAggregate>() {
    private Connection conn;
    private PreparedStatement stmt;

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = DriverManager.getConnection(DB_URL, DB_USER, DB_PASS);
        stmt = conn.prepareStatement(
            "INSERT INTO user_risk_stats (user_id, total_amount, window_end) VALUES (?, ?, ?)");
    }

    @Override
    public void invoke(UserAggregate value, Context context) throws Exception {
        stmt.setString(1, value.getUserId());
        stmt.setDouble(2, value.getTotalAmount());
        stmt.setLong(3, value.getWindowEnd());
        stmt.executeUpdate();
    }

    @Override
    public void close() throws Exception {
        if (stmt != null) stmt.close();
        if (conn != null) conn.close();
    }
});

3.5 完整风控管道示例

以下 SVG 图展示了从 Kafka Source 到 Sink 的完整数据流，以及各 Transformation 算子在风控巡检中的位置：

3.6 算子选择避坑指南