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

在贷后预警系统中，数据从产生到最终落库或触发告警，需要经过完整的 ETL 管道。本章聚焦 Flink DataStream API 的三大核心环节：Source（数据源）、Transformation（转换） 与 Sink（数据输出）。我们将以贷后预警场景为例，实战演练 Kafka 实时流接入、MySQL CDC 变更捕获、核心算子链式处理，以及结果写入 Kafka 和 MySQL JDBC 的完整流程。

实战目标：掌握 Flink DataStream API 中 Source、Transformation、Sink 的算子选择与链式编排，能够独立搭建一条从 Kafka/MySQL CDC 到预警结果输出的实时处理管道。

3.1 数据源（Source）实战

贷后预警系统需要同时处理实时事件流与数据库变更流。Flink 提供了丰富的 Source 连接器，本节聚焦最常用的两种：Kafka Source 与 MySQL CDC Source。

3.1.1 Kafka Source：实时事件流接入

Kafka 是贷后预警系统接收还款行为、逾期状态变更等实时事件的核心通道。使用 Flink Kafka Connector 可以精确一次（Exactly-Once）地消费 Kafka 主题。

// 构建 Kafka Source
DataStream<String> kafkaStream = env.fromSource(
    KafkaSource.<String>builder()
        .setBootstrapServers("localhost:9092")
        .setTopics("loan-repayment-events")
        .setGroupId("flink-warning-group")
        .setStartingOffsets(OffsetsInitializer.latest())
        .setValueOnlyDeserializer(new SimpleStringSchema())
        .build(),
    WatermarkStrategy.noWatermarks(),
    "Kafka Source"
);
配置项 说明 贷后预警推荐值
setBootstrapServers Kafka 集群地址 生产环境配置多个 Broker 地址
setTopics 订阅的主题名 按事件类型分主题（如还款、逾期、调额）
setGroupId 消费者组 ID 每个预警作业使用独立 Group ID
setStartingOffsets 起始消费位置 首次启动用 earliest，重启用 latest

3.1.2 MySQL CDC Source：数据库变更捕获

贷后预警需要实时感知数据库中的客户信息、合同状态、还款计划等变更。Flink CDC（Change Data Capture）基于 Debezium 引擎，能够将 MySQL Binlog 实时解析为变更事件流。

// 构建 MySQL CDC Source
DataStream<RowData> cdcStream = env.fromSource(
    MySQLSource.<RowData>builder()
        .hostname("localhost")
        .port(3306)
        .databaseList("loan_db")
        .tableList("loan_db.contract_info", "loan_db.repayment_plan")
        .username("cdc_user")
        .password("cdc_pass")
        .deserializer(new JsonDebeziumDeserializationSchema())
        .startupOptions(StartupOptions.initial())
        .build(),
    WatermarkStrategy.noWatermarks(),
    "MySQL CDC Source"
);
关键说明：CDC Source 的 startupOptions.initial() 会先做一次全量快照，再无缝切换到增量 Binlog 监听，非常适合贷后预警系统初始化时加载存量数据。

3.2 转换（Transformation）核心算子

Flink DataStream API 提供了丰富的转换算子，用于对数据流进行过滤、映射、分组等操作。贷后预警场景中最常用的三个算子是 map、filter 和 keyBy。

3.2.1 map：数据格式转换与字段提取

map 算子对数据流中的每个元素执行一对一转换。在贷后预警中，常用于将 JSON 字符串解析为 Java POJO，或提取关键字段。

// 将 JSON 字符串解析为 RepaymentEvent 对象
DataStream<RepaymentEvent> eventStream = kafkaStream
    .map(json -> JsonUtil.parseObject(json, RepaymentEvent.class));

3.2.2 filter：条件过滤与数据清洗

filter 算子根据布尔条件保留或丢弃元素。贷后预警中，常用于过滤掉无效数据或只保留特定类型的事件。

// 只保留逾期天数大于 0 的逾期事件
DataStream<OverdueEvent> overdueStream = eventStream
    .filter(event -> event.getOverdueDays() > 0);

3.2.3 keyBy：按业务键分组

keyBy 算子根据指定的 Key 对数据流进行逻辑分区，后续的窗口、状态操作都基于分区执行。贷后预警中，通常按 合同号 或 客户 ID 进行分组。

// 按合同号分组，用于后续计算每个合同的累计逾期天数
KeyedStream<OverdueEvent, String> keyedStream = overdueStream
    .keyBy(OverdueEvent::getContractId);
避坑提示：keyBy 后的数据分布取决于 Key 的哈希值。如果 Key 分布不均匀（如热点合同），会导致数据倾斜。建议在 Key 中拼接随机后缀或使用 rebalance() 算子进行负载均衡。

3.3 数据输出（Sink）实战

贷后预警结果需要同时写入多个目标系统：实时告警消息发送到 Kafka 主题，持久化预警记录写入 MySQL 数据库。Flink 提供了 Kafka Sink 和 JDBC Sink 两种主流输出方式。

3.3.1 Kafka Sink：实时告警消息输出

将预警结果发送到 Kafka 告警主题，下游的告警服务或通知系统可以实时消费。

// 构建 Kafka Sink
warningStream.sinkTo(
    KafkaSink.<WarningResult>builder()
        .setBootstrapServers("localhost:9092")
        .setRecordSerializer(KafkaRecordSerializationSchema.builder()
            .setTopic("loan-warning-results")
            .setValueSerializationSchema(new JsonSerializationSchema<>())
            .build()
        )
        .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
        .build()
);

3.3.2 MySQL JDBC Sink：预警记录持久化

使用 Flink JDBC Connector 将预警结果写入 MySQL 表，便于后续查询、报表和审计。

// 构建 JDBC Sink
warningStream.addSink(
    JdbcSink.sink(
        "INSERT INTO warning_record(contract_id, customer_id, warning_type, warning_level, create_time) VALUES (?, ?, ?, ?, ?)",
        (statement, record) -> {
            statement.setString(1, record.getContractId());
            statement.setString(2, record.getCustomerId());
            statement.setString(3, record.getWarningType());
            statement.setString(4, record.getWarningLevel());
            statement.setTimestamp(5, record.getCreateTime());
        },
        JdbcExecutionOptions.builder()
            .withBatchSize(100)
            .withBatchIntervalMs(200)
            .build(),
        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
            .withUrl("jdbc:mysql://localhost:3306/warning_db")
            .withUsername("flink_user")
            .withPassword("flink_pass")
            .build()
    )
);
Sink 类型 适用场景 一致性保证 贷后预警用途
Kafka Sink 实时消息推送 至少一次 / 精确一次 告警通知、下游系统联动
MySQL JDBC Sink 持久化存储 至少一次（幂等写入） 预警记录归档、报表查询

3.4 贷后预警管道完整链路

将上述 Source、Transformation、Sink 串联起来，形成一条完整的贷后预警实时处理管道。下面的 SVG 图展示了数据从 Kafka 和 MySQL CDC 流入，经过核心算子处理，最终输出到 Kafka 和 MySQL 的全流程。

贷后预警 DataStream API 管道架构图 贷后预警 DataStream API 管道架构图 Kafka Source 还款事件 / 逾期事件 MySQL CDC Source 合同 / 还款计划变更 Transformation 核心算子链 map JSON → POJO filter 逾期天数 > 0 keyBy 按合同号分组 process 预警规则评估 Kafka Sink 告警消息主题 MySQL JDBC Sink 预警记录表 双路输出策略 实时告警 + 持久化存储

3.5 完整代码示例：贷后预警管道

以下代码将上述所有环节整合为一个完整的 Flink 作业，实现从 Kafka 消费还款事件、过滤逾期记录、按合同分组、评估预警规则，最终将结果同时写入 Kafka 和 MySQL。

public class LoanWarningPipeline {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        
        // 1. Kafka Source：消费还款事件
        DataStream<String> kafkaSource = env.fromSource(
            KafkaSource.<String>builder()
                .setBootstrapServers("localhost:9092")
                .setTopics("loan-repayment-events")
                .setGroupId("flink-warning-group")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build(),
            WatermarkStrategy.noWatermarks(),
            "Kafka Source"
        );
        
        // 2. map：JSON 解析为 POJO
        DataStream<RepaymentEvent> eventStream = kafkaSource
            .map(json -> JsonUtil.parseObject(json, RepaymentEvent.class));
        
        // 3. filter：只保留逾期事件
        DataStream<OverdueEvent> overdueStream = eventStream
            .filter(event -> event.getOverdueDays() > 0);
        
        // 4. keyBy：按合同号分组
        KeyedStream<OverdueEvent, String> keyedStream = overdueStream
            .keyBy(OverdueEvent::getContractId);
        
        // 5. process：预警规则评估（状态 + 定时器）
        DataStream<WarningResult> warningStream = keyedStream
            .process(new WarningRuleProcessFunction());
        
        // 6. Kafka Sink：发送告警消息
        warningStream.sinkTo(
            KafkaSink.<WarningResult>builder()
                .setBootstrapServers("localhost:9092")
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                    .setTopic("loan-warning-results")
                    .setValueSerializationSchema(new JsonSerializationSchema<>())
                    .build()
                )
                .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
                .build()
        );
        
        // 7. MySQL JDBC Sink：持久化预警记录
        warningStream.addSink(
            JdbcSink.sink(
                "INSERT INTO warning_record(contract_id, customer_id, warning_type, warning_level, create_time) VALUES (?, ?, ?, ?, ?)",
                (statement, record) -> {
                    statement.setString(1, record.getContractId());
                    statement.setString(2, record.getCustomerId());
                    statement.setString(3, record.getWarningType());
                    statement.setString(4, record.getWarningLevel());
                    statement.setTimestamp(5, record.getCreateTime());
                },
                JdbcExecutionOptions.builder()
                    .withBatchSize(100)
                    .withBatchIntervalMs(200)
                    .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                    .withUrl("jdbc:mysql://localhost:3306/warning_db")
                    .withUsername("flink_user")
                    .withPassword("flink_pass")
                    .build()
            )
        );
        
        env.execute("Loan Warning Pipeline");
    }
}
最佳实践：生产环境中建议将 Kafka Source 的并行度与 Kafka 分区数对齐，JDBC Sink 使用 withBatchSize 和 withBatchIntervalMs 控制写入频率，避免频繁数据库连接开销。

3.6 本章小结

通过本章实战，你已掌握 Flink DataStream API 中 Source、Transformation、Sink 三大核心环节的选型与编码方法。在贷后预警场景中：

  • Source 层：Kafka Source 接入实时事件流，MySQL CDC Source 捕获数据库变更，两者互补覆盖全量数据源。
  • Transformation 层：map 做格式转换，filter 做条件过滤，keyBy 做业务分组，process 做复杂状态计算。
  • Sink 层：Kafka Sink 实现实时告警推送，MySQL JDBC Sink 实现持久化存储，双路输出保障数据不丢失。
← 上一章 📖 返回目录 下一章 →