📚 课程目录第4章 Flink DataStream API入门

4. Flink DataStream API 入门：Source、Transformation、Sink

在精算坏样本归因场景中，Flink 实时计算的核心是构建一条从数据接入、逻辑处理到结果输出的完整管道。本章将系统讲解 DataStream API 的三大核心组件：Source（数据源）、Transformation（转换算子） 和 Sink（数据汇），并结合 Kafka、文件、MySQL、Elasticsearch 等实际组件，演示如何在精算链路中落地。

4.1 整体架构概览

下图展示了 Flink DataStream 作业在精算坏样本归因中的标准数据流拓扑。数据从 Kafka 或文件系统接入，经过多级转换后，最终写入 MySQL（存储归因结果明细）和 Elasticsearch（提供实时检索与聚合）。

4.2 Source：数据接入

Source 是 DataStream 的起点，负责从外部系统读取数据并转换为 DataStream<T>。精算场景中，最常用的 Source 是 Kafka（实时事件流）和文件（历史批次/补录数据）。

4.2.1 Kafka Source

Kafka 作为高吞吐、低延迟的消息队列，是实时归因的首选数据总线。Flink 通过 FlinkKafkaConsumer 或新的 KafkaSource API 接入。

  精算场景示例： 从 Kafka 主题 actuary-claim-events 读取理赔事件，每条消息为 JSON 格式，包含保单号、出险时间、损失金额等字段。

// 使用 KafkaSource API（推荐）
KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("kafka-broker:9092")
    .setTopics("actuary-claim-events")
    .setGroupId("flink-actuary-group")
    .setStartingOffsets(OffsetsInitializer.latest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

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

配置项	说明	精算推荐值
`bootstrapServers`	Kafka 集群地址	生产环境至少 3 节点
`topics`	订阅的主题名	按险种/业务线隔离
`groupId`	消费者组 ID	保证作业重启后 offset 不丢失
`startingOffsets`	起始消费位置	精算归因建议 `latest` 或 `timestamp`

4.2.2 File Source

文件 Source 适用于读取历史坏样本数据（如 CSV、Parquet 格式），用于模型初始化或离线补录。Flink 支持一次性读取或持续监控目录。

// 读取历史坏样本 CSV 文件
DataStream<String> historyStream = env.readTextFile("hdfs:///data/actuary/bad_samples/2024/");

// 持续监控目录（新文件自动读取）
DataStream<String> monitorStream = env
    .readFile(new TextInputFormat(new Path("hdfs:///data/actuary/bad_samples/incoming/")),
              "hdfs:///data/actuary/bad_samples/incoming/",
              FileProcessingMode.PROCESS_CONTINUOUSLY, 10000);

最佳实践： 精算历史数据通常较大，建议使用 readFile 配合 PROCESS_CONTINUOUSLY 模式，实现准实时补录。同时注意设置合理的监控间隔（如 10 秒），避免频繁扫描 HDFS。

4.3 Transformation：核心转换算子

Transformation 是 Flink 作业的逻辑核心。精算坏样本归因中，最常用的三个算子为 map、filter 和 keyBy。它们共同完成数据清洗、特征提取和分组聚合。

4.3.1 map：数据映射与转换

map 算子对每条记录执行一对一转换，常用于 JSON 解析、字段提取、数据增强。

// 将 JSON 字符串解析为 ClaimEvent 对象
DataStream<ClaimEvent> parsedStream = claimStream
    .map(new MapFunction<String, ClaimEvent>() {
        @Override
        public ClaimEvent map(String value) throws Exception {
            ObjectMapper mapper = new ObjectMapper();
            return mapper.readValue(value, ClaimEvent.class);
        }
    });

// 精算特征增强：计算损失率
DataStream<ClaimEvent> enrichedStream = parsedStream
    .map(event -> {
        event.setLossRatio(event.getLossAmount() / event.getPremium());
        return event;
    });

4.3.2 filter：数据过滤

filter 算子根据条件保留或丢弃记录。精算场景中，常用于剔除无效数据、筛选特定险种或损失范围。

// 过滤出损失金额大于 0 且保单状态为有效的理赔事件
DataStream<ClaimEvent> validStream = enrichedStream
    .filter(event -> event.getLossAmount() > 0
            && "ACTIVE".equals(event.getPolicyStatus()));

// 筛选车险业务线
DataStream<ClaimEvent> autoStream = validStream
    .filter(event -> "AUTO".equals(event.getProductLine()));

注意： filter 算子不会改变数据类型，但会显著影响下游数据量。精算归因中，建议在 map 之后尽早执行 filter，减少后续算子的处理压力。

4.3.3 keyBy：分组与分区

keyBy 算子根据指定 key 对数据流进行逻辑分组，是后续聚合操作（如 sum、reduce、window）的前提。精算归因中，通常按保单号、客户 ID 或险种分组。

// 按保单号分组，用于后续归因聚合
KeyedStream<ClaimEvent, String> keyedStream = validStream
    .keyBy(ClaimEvent::getPolicyNo);

// 按险种分组，用于统计各险种损失分布
KeyedStream<ClaimEvent, String> productKeyed = validStream
    .keyBy(ClaimEvent::getProductLine);

算子	输入 → 输出	精算典型用途	注意事项
`map`	1 条 → 1 条	JSON 解析、特征计算、字段映射	避免在 map 中执行耗时 I/O
`filter`	1 条 → 0 或 1 条	剔除无效保单、筛选险种	尽早执行，减少下游数据量
`keyBy`	DataStream → KeyedStream	按保单号/客户 ID 分组	key 的 hashCode 需均匀分布

4.4 Sink：结果输出

Sink 是 DataStream 的终点，负责将处理结果写入外部存储。精算坏样本归因中，MySQL 用于存储归因结果明细（支持事务性查询），Elasticsearch 用于实时检索与聚合分析。

4.4.1 MySQL Sink

Flink 通过 JDBC Connector 或自定义 Sink 写入 MySQL。推荐使用 JdbcSink 或 JdbcExecutionOptions 控制批量写入行为。

// 定义 MySQL Sink，写入归因结果表
DataStream<AttributionResult> resultStream = ...; // 上游归因结果

resultStream.addSink(
    JdbcSink.sink(
        "INSERT INTO attribution_result (policy_no, loss_amount, attribution_score, event_time) VALUES (?, ?, ?, ?)",
        (statement, result) -> {
            statement.setString(1, result.getPolicyNo());
            statement.setDouble(2, result.getLossAmount());
            statement.setDouble(3, result.getAttributionScore());
            statement.setTimestamp(4, result.getEventTime());
        },
        JdbcExecutionOptions.builder()
            .withBatchSize(1000)
            .withBatchIntervalMs(5000)
            .build(),
        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
            .withUrl("jdbc:mysql://mysql-host:3306/actuary_db")
            .withDriverName("com.mysql.cj.jdbc.Driver")
            .withUsername("flink_user")
            .withPassword("flink_pass")
            .build()
    )
);

精算优化建议： 设置 batchSize 为 500~1000，batchIntervalMs 为 3~5 秒，在吞吐量和实时性之间取得平衡。对于高精度归因场景，可考虑使用 JdbcExactlyOnceSink 保证精确一次语义。

4.4.2 Elasticsearch Sink

Elasticsearch Sink 支持将归因结果索引化，便于精算师通过 Kibana 进行实时探查和聚合分析。

// 定义 Elasticsearch Sink
ElasticsearchSink<AttributionResult> esSink = ElasticsearchSink.<AttributionResult>builder()
    .setHosts(new HttpHost("es-node1", 9200, "http"))
    .setBulkFlushMaxActions(500)
    .setBulkFlushInterval(3000)
    .setConnectionRequestTimeout(5000)
    .setFailureHandler(new RetryRejectedExecutionFailureHandler())
    .build();

resultStream.sinkTo(esSink);

配置项	说明	精算推荐值
`bulkFlushMaxActions`	批量写入最大条数	500~1000
`bulkFlushInterval`	批量写入最大间隔（ms）	3000~5000
`failureHandler`	写入失败处理策略	建议使用 `RetryRejectedExecutionFailureHandler`

4.5 完整作业示例：精算归因管道

以下代码整合了本章所有知识点，构建一个完整的精算坏样本归因 Flink 作业：从 Kafka 读取事件，经过 map/filter/keyBy 转换，最终写入 MySQL 和 Elasticsearch。

public class ActuaryAttributionJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1. Source：Kafka 接入
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("actuary-claim-events")
            .setGroupId("flink-actuary")
            .setStartingOffsets(OffsetsInitializer.latest())
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();

        DataStream<String> rawStream = env.fromSource(kafkaSource,
            WatermarkStrategy.noWatermarks(), "Kafka Source");

        // 2. Transformation：解析 → 过滤 → 分组
        DataStream<ClaimEvent> parsed = rawStream
            .map(value -> new ObjectMapper().readValue(value, ClaimEvent.class));

        DataStream<ClaimEvent> valid = parsed
            .filter(e -> e.getLossAmount() > 0 && "ACTIVE".equals(e.getPolicyStatus()));

        KeyedStream<ClaimEvent, String> keyed = valid
            .keyBy(ClaimEvent::getPolicyNo);

        // 3. 归因逻辑（示例：计算归因分数）
        DataStream<AttributionResult> results = keyed
            .map(event -> {
                double score = event.getLossAmount() / event.getPremium() * 100;
                return new AttributionResult(event.getPolicyNo(),
                    event.getLossAmount(), score, event.getEventTime());
            });

        // 4. Sink：MySQL + Elasticsearch
        results.addSink(buildMySqlSink());
        results.sinkTo(buildEsSink());

        env.execute("Actuary Attribution Job");
    }
}

4.6 本章小结

Source：Kafka 适合实时事件流，File 适合历史数据补录；注意 offset 管理和监控间隔。
Transformation：map 做一对一转换，filter 做条件过滤，keyBy 做分组；三者组合可完成大部分精算预处理逻辑。
Sink：MySQL 适合事务性结果存储，Elasticsearch 适合实时检索；合理配置批量参数可大幅提升吞吐量。
作业拓扑：Source → Transformation → Sink 构成标准管道，每个环节的并行度和分区策略直接影响作业性能。

  课后练习： 尝试修改上述完整作业，增加一个 filter 步骤，仅保留损失金额大于 10000 元的理赔事件，并观察 MySQL 和 ES 中数据量的变化。