4. Flink DataStream API 入门：Source、Transformation、Sink 核心算子详解

在交易风控场景中，实时数据流从产生到最终输出告警或阻断指令，需要经过完整的 Source → Transformation → Sink 三阶段处理。Flink DataStream API 提供了丰富的算子来支撑这一链路。本章将围绕风控样本切分的实际需求，深入解析三类核心算子的原理与实战用法。

4.1 整体架构与数据流

下图展示了 Flink DataStream API 在交易风控样本切分中的完整数据流架构。从实时交易事件接入，经过多级 Transformation 处理，最终输出切分后的样本集与风控指标。

4.2 Source 算子详解

Source 是数据流的起点。Flink 内置了多种 Source 实现，同时也支持自定义 Source。在交易风控场景中，最常用的 Source 是 Kafka Source 和 Socket Source（用于测试）。

4.2.1 常用 Source 对比

4.2.2 Kafka Source 实战代码

以下代码展示了如何在 Flink 中创建 Kafka Source，并反序列化交易事件为 Java POJO。

// 1. 设置 Kafka 连接属性
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
kafkaProps.setProperty("group.id", "risk_control_group");

// 2. 创建 Kafka Source（Flink 1.12+ 推荐新 API）
DataStream<String> rawStream = env.fromSource(
    KafkaSource.<String>builder()
        .setBootstrapServers("localhost:9092")
        .setTopics("trade-events")
        .setGroupId("risk_control_group")
        .setStartingOffsets(OffsetsInitializer.latest())
        .setValueOnlyDeserializer(new SimpleStringSchema())
        .build(),
    WatermarkStrategy.noWatermarks(),
    "Kafka Source"
);

// 3. 将 JSON 字符串解析为交易事件对象
DataStream<TradeEvent> tradeStream = rawStream
    .map(json -> JsonUtil.parse(json, TradeEvent.class));

4.3 Transformation 核心算子

Transformation 是 Flink 数据处理的核心。在交易风控样本切分中，以下算子最为关键：

4.3.1 基础转换算子

4.3.2 风控样本切分中的 Transformation 链

以下代码展示了从原始交易流到样本切分的完整 Transformation 链路：

// 1. 过滤：只保留有效交易（状态码为 SUCCESS）
DataStream<TradeEvent> validTrades = tradeStream
    .filter(trade -> "SUCCESS".equals(trade.getStatus()));

// 2. 标记样本标签：根据风控规则打标（1=欺诈，0=正常）
DataStream<SampleRecord> sampleStream = validTrades
    .map(trade -> {
        int label = trade.getAmount() > 50000 ? 1 : 0;  // 简化规则
        return new SampleRecord(trade.getUserId(), trade.getAmount(),
                                trade.getTimestamp(), label);
    });

// 3. 按用户分组，统计窗口内交易频次
DataStream<UserFeature> featureStream = sampleStream
    .keyBy(SampleRecord::getUserId)
    .window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
    .aggregate(new TradeCountAggregate())
    .name("user-trade-count-window");

// 4. 将样本与特征关联，生成最终训练样本
DataStream<TrainingSample> trainingSamples = sampleStream
    .connect(featureStream)
    .keyBy(SampleRecord::getUserId, UserFeature::getUserId)
    .process(new SampleJoinFunction());

4.3.3 窗口算子详解

窗口是 Flink 处理时间序列数据的关键。在风控样本切分中，常用窗口类型如下：

4.4 Sink 算子详解

Sink 是数据流的终点。在交易风控样本切分中，Sink 负责将处理后的样本数据输出到外部存储系统。

4.4.1 常用 Sink 对比

4.4.2 Kafka Sink 实战代码

// 1. 创建 Kafka Sink（Flink 1.15+ 推荐新 API）
KafkaSink<TrainingSample> kafkaSink = KafkaSink.<TrainingSample>builder()
    .setBootstrapServers("localhost:9092")
    .setRecordSerializer(
        KafkaRecordSerializationSchema.builder()
            .setTopic("risk-samples")
            .setValueSerializationSchema(
                new JsonSerializationSchema<>(
                    () -> TrainingSample.class))
            .build()
    )
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .build();

// 2. 将样本流写入 Kafka
trainingSamples.sinkTo(kafkaSink).name("risk-sample-kafka-sink");

// 3. 同时将样本写入 MySQL（双写策略）
trainingSamples.addSink(
    JdbcSink.sink(
        "INSERT INTO risk_samples(user_id, amount, label, feature_json) VALUES (?, ?, ?, ?)",
        (statement, sample) -> {
            statement.setString(1, sample.getUserId());
            statement.setDouble(2, sample.getAmount());
            statement.setInt(3, sample.getLabel());
            statement.setString(4, sample.getFeatureJson());
        },
        JdbcExecutionOptions.builder()
            .withBatchSize(100)
            .withBatchIntervalMs(200)
            .build(),
        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
            .withUrl("jdbc:mysql://localhost:3306/risk_db")
            .withDriverName("com.mysql.cj.jdbc.Driver")
            .withUsername("risk_user")
            .withPassword("risk_pass")
            .build()
    )
).name("risk-sample-mysql-sink");

4.5 算子链与性能优化

在风控样本切分作业中，合理的算子链配置可以显著提升吞吐量。Flink 默认会将满足条件的算子合并到同一个 Task 中执行，减少序列化/反序列化和网络开销。

4.5.1 算子链优化策略

• 默认合并：map、filter、flatMap 等无状态算子默认会链在一起。
• 禁止链：keyBy、window、connect 等需要网络 Shuffle 的算子会打断链。
• 手动控制：使用 .disableChaining() 强制断开链，或 .startNewChain() 开启新链。

// 示例：手动控制算子链
DataStream<TradeEvent> stream = env
    .fromSource(kafkaSource, ...)
    .map(new ParseFunction()).disableChaining()  // 强制断开，便于调试
    .filter(new HighValueFilter()).startNewChain() // 开启新链
    .keyBy(TradeEvent::getUserId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new TradeAggregate());

4.5.2 并行度配置建议

4.6 本章小结

• Source 是数据流的入口，Kafka Source 是风控场景的首选，支持 Exactly-once 语义。
• Transformation 是数据处理的核心，map/filter/keyBy/window 的组合使用可以完成从原始交易到样本切分的全流程。
• Sink 是数据流的出口，Kafka Sink 和 JDBC Sink 分别适用于实时输出和持久化存储。
• 算子链 和 并行度 的合理配置是保障风控作业高性能运行的关键。