简介：这是一份面向大数据开发初学者与进阶工程师的Flink实战项目资料，基于保险行业真实业务场景，采用Flink+HBase+Kafka+Phoenix架构，实现业务系统数据库数据的实时同步与实时统计报表分析，适合想通过完整项目理解流式计算落地流程的读者。资源包共73个文件，约584KB，以34个Java源码为核心，辅以11个Python脚本、9个Shell脚本，以及csv、properties、xml、sql、jar等配置与数据文件，并附有说明文档，覆盖从数据采集、处理到存储查询的完整链路。目前已有1166人学习下载。通过该资料，读者可掌握实时数仓的模块划分与代码组织方式，理解Kafka接入、Flink计算、HBase与Phoenix存储查询的协作逻辑，并借助脚本与配置快速搭建本地运行环境，同时作者提供Flink答疑服务，便于快速入门与排查问题。

1. 保险核心系统实时化：为什么 Flink 成了那个绕不开的选项

保险行业的核心系统有个特点：白天跑批、晚上对账、月底出报表。这个节奏在过去二十年没什么问题，直到业务方开始要求「保全申请提交后 3 秒内看到核保结论」「理赔报案后实时判断是否触发反欺诈规则」「渠道佣金按小时结算」。这些需求落到技术侧，本质是把原来 T+1 的批量计算压缩到秒级甚至毫秒级，而且数据源横跨 Oracle 核心库、MySQL 渠道库、Kafka 埋点流和文件批量导入。

Flink 在这个场景里被反复选中，原因不复杂：它能同时处理「有界的历史数据」和「无界的实时流」，Exactly-Once 语义在保险这种对金额极度敏感的场景里是刚需，窗口和水位线机制天然适配「按保单维度做时间窗口聚合」这类操作。保险行业的 Flink 实战项目，核心不是把 WordCount 跑通，而是解决三个具体问题：保单主数据变更如何实时捕获、核保规则如何在不重启作业的前提下动态更新、理赔金额聚合如何保证不重不漏。这篇内容按「数据接入 → 规则计算 → 状态管理 → 生产避坑」的路径展开，适合已经了解 Flink 基础 API、准备在保险或金融场景落地实时计算的工程师。

2. 保险数据接入层：从 Oracle CDC 到 Kafka 的完整链路

2.1 为什么保险核心库的 CDC 不能直接用 Flink CDC 默认配置

保险核心库通常是 Oracle，表结构有几个典型特征：保单表按年份分表（POLICY_2023、POLICY_2024）、字段多且存在大量 CLOB 备注字段、更新操作集中在特定时段（比如每天 20:00 后的批量保全）。直接用 Flink CDC 的 Oracle Connector 默认配置会踩三个坑：第一，分表场景下需要手动维护表名列表，新增年份分表时作业要重启；第二，CLOB 字段默认会被忽略或截断，导致备注信息丢失；第三，大批量更新时 redo log 暴涨，DBA 会来找你。

常见做法是用 Debezium 做 Oracle 的 LogMiner 采集，输出到 Kafka 后由 Flink 消费。这样做的理由是：Debezium 对 Oracle LogMiner 的适配更成熟，支持按 schema 动态发现新表，CLOB 字段可以通过配置decimal.handling.mode和column.propagate.source.type保留完整内容。Flink 侧只负责消费 Kafka 并做后续计算，职责分离后排查问题也清晰——数据没到 Flink，查 Debezium；到了 Flink 算错了，查作业逻辑。

# Debezium Oracle Connector 关键配置（Kafka Connect 格式） { "name": "insurance-policy-cdc", "config": { "connector.class": "io.debezium.connector.oracle.OracleConnector", "database.hostname": "core-db-insurance", "database.port": "1521", "database.user": "cdc_user", "database.password": "******", "database.dbname": "CORE", "database.pdb.name": "PDB_PROD", "table.include.list": "POLICY\\.POLICY_.*", "schema.history.internal.kafka.bootstrap.servers": "kafka-broker:9092", "schema.history.internal.kafka.topic": "schema-changes.policy", "log.mining.strategy": "online_catalog", "log.mining.continuous.mine": "true", "decimal.handling.mode": "string", "column.propagate.source.type": "POLICY\\.POLICY_.*", "snapshot.mode": "initial", "topic.prefix": "insurance-cdc" } }

这段配置里几个参数需要重点说明。table.include.list用正则匹配所有年份分表，新增 POLICY_2025 时不需要改配置。log.mining.strategy设为online_catalog而不是默认的redo_log_catalog，前者对在线字典的读取更稳定，后者在 DDL 变更后容易报错。decimal.handling.mode设为string是为了避免金额字段在 JSON 序列化时丢失精度——保险场景里 0.01 元的误差都可能导致对账失败。snapshot.mode设为initial表示首次启动时先做全量快照再增量，如果表数据量超过千万级，建议改为schema_only并单独用 DataX 做历史数据初始化。

2.2 Flink 消费 Kafka 时的反序列化与水位线设置

Debezium 输出的 Kafka 消息格式是 JSON，包含before、after、op、ts_ms四个核心字段。Flink 侧需要自定义DeserializationSchema把 JSON 转成RowData或 POJO。这里有个容易翻车的地方：Debezium 的ts_ms是数据库变更时间，但 Kafka 消息的 timestamp 是写入时间，两者可能相差几秒到几分钟。如果直接用水位线基于 Kafka timestamp，窗口触发会延迟；如果用ts_ms，需要处理乱序问题。

// Flink Kafka Source 配置，基于 Debezium JSON 格式 KafkaSource<PolicyChangeEvent> source = KafkaSource.<PolicyChangeEvent>builder() .setBootstrapServers("kafka-broker:9092") .setTopics("insurance-cdc.POLICY.POLICY_2024") .setGroupId("flink-policy-consumer") .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST)) .setDeserializer(new DebeziumJsonDeserializer()) .setProperty("partition.discovery.interval.ms", "60000") .build(); DataStream<PolicyChangeEvent> stream = env.fromSource( source, WatermarkStrategy.<PolicyChangeEvent>forBoundedOutOfOrderness(Duration.ofSeconds(30)) .withTimestampAssigner((event, recordTs) -> event.getTsMs()) .withIdleness(Duration.ofMinutes(1)), "PolicyCDC" );

forBoundedOutOfOrderness(Duration.ofSeconds(30))表示允许 30 秒的乱序，这个值根据实际观察到的ts_ms与处理时间的差值来定。保险核心库在批量保全时段，变更事件可能集中爆发，30 秒是保守估计。withIdleness(Duration.ofMinutes(1))解决的是分表场景下某个分区长时间无数据导致水位线不推进的问题——比如 POLICY_2023 在 2024 年已经很少变更，如果没有 idleness 检测，整个作业的水位线会被这个空闲分区拖住。

partition.discovery.interval.ms设为 60000 表示每分钟检查一次新分区。当 Debezium 发现新表并创建新 topic 分区时，Flink 能自动感知。但注意，这个参数只对 Kafka Source 的分区发现有效，如果 Debezium 新增了完全不同的 topic，还是需要重启作业或使用topic-pattern模式。

3. 核保规则计算：广播状态与动态规则更新

3.1 用 BroadcastState 实现不重启作业更新核保规则

保险核保规则的特点是频繁调整：今天银保渠道的免体检额度是 50 万，明天可能调到 80 万；某个职业类别今天在拒保名单里，明天可能移出。如果每次规则变更都重启 Flink 作业，不仅影响实时性，还会导致状态丢失和重复计算。BroadcastState 是 Flink 提供的解决方案：一条流是保单变更事件（主流），另一条流是规则更新事件（广播流），广播流的数据会被分发到主流的所有并行实例上。

// 广播状态描述符：存储规则 ID -> 规则内容的映射 MapStateDescriptor<String, UnderwritingRule> ruleStateDesc = new MapStateDescriptor<>( "underwriting-rules", BasicTypeInfo.STRING_TYPE_INFO, TypeInformation.of(UnderwritingRule.class) ); // 规则流：从 Kafka 或配置中心读取规则变更 DataStream<UnderwritingRule> ruleStream = env .addSource(new FlinkKafkaConsumer<>("underwriting-rules", new RuleDeserializer(), props)) .broadcast(ruleStateDesc); // 主流：保单变更事件 DataStream<PolicyChangeEvent> policyStream = env .addSource(new FlinkKafkaConsumer<>("insurance-cdc.POLICY.POLICY_2024", new DebeziumJsonDeserializer(), props)) .keyBy(PolicyChangeEvent::getPolicyNo); // 连接广播流 BroadcastConnectedStream<PolicyChangeEvent, UnderwritingRule> connected = policyStream.connect(ruleStream); connected.process(new BroadcastProcessFunction<PolicyChangeEvent, UnderwritingRule, UnderwritingResult>() { @Override public void processElement(PolicyChangeEvent event, ReadOnlyContext ctx, Collector<UnderwritingResult> out) throws Exception { // 从广播状态中读取当前生效的规则 MapState<String, UnderwritingRule> rules = ctx.getBroadcastState(ruleStateDesc).immutableEntries(); UnderwritingRule rule = rules.get(event.getChannelCode() + "_" + event.getProductCode()); if (rule == null) { // 规则未配置时走默认兜底逻辑 out.collect(UnderwritingResult.defaultPass(event)); return; } // 执行核保判断 UnderwritingResult result = rule.evaluate(event); out.collect(result); } @Override public void processBroadcastElement(UnderwritingRule rule, Context ctx, Collector<UnderwritingResult> out) throws Exception { // 规则更新时写入广播状态 ctx.getBroadcastState(ruleStateDesc).put(rule.getKey(), rule); } });

processElement里每次从广播状态读取规则，而不是缓存到成员变量，原因是广播状态在 checkpoint 时会持久化，作业恢复后规则不会丢失。processBroadcastElement里直接put覆盖，保证规则更新后立即生效。注意immutableEntries()返回的是只读视图，不要尝试修改。

一个实际踩过的坑：广播状态默认存储在堆内存中，如果规则数量超过几千条且每条规则包含复杂条件表达式，内存压力会很大。我一般会把规则做两级拆分——高频变化的阈值类规则放广播状态，低频变化的复杂规则表达式放外部缓存（如 Redis），广播流只传递规则版本号，主流根据版本号去 Redis 拉取完整规则。

3.2 核保结果的对账与 Exactly-Once 保证

保险场景对金额和结论的准确性要求极高，核保结果写回下游系统时必须保证 Exactly-Once。Flink 的 checkpoint 机制配合两阶段提交（2PC）可以实现端到端的一致性，但需要下游 Sink 支持事务。如果下游是 MySQL，可以用JdbcSink配合XADataSource；如果下游是 Kafka，用KafkaSink的EXACTLY_ONCE模式。

// Kafka Sink 的 Exactly-Once 配置 KafkaSink<UnderwritingResult> sink = KafkaSink.<UnderwritingResult>builder() .setBootstrapServers("kafka-broker:9092") .setRecordSerializer(new UnderwritingResultSerializer("underwriting-result")) .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) .setTransactionalIdPrefix("flink-underwriting-") .setProperty("transaction.timeout.ms", "900000") .build(); resultStream.sinkTo(sink);

transaction.timeout.ms设为 900000（15 分钟）是因为保险核保作业的 checkpoint 间隔通常设为 5 分钟，事务超时时间需要大于 checkpoint 间隔的两倍以上，否则事务可能在 checkpoint 完成前超时回滚。setTransactionalIdPrefix必须保证全局唯一，否则多个作业实例会互相干扰。

对账逻辑建议单独做一个离线校验作业：每天凌晨读取前一天 Kafka 中的核保结果和 MySQL 中的核保记录，按保单号做全量比对，发现不一致时告警。这个校验作业不需要 Flink，用 Spark 或直接 SQL 都能做，关键是形成闭环。

4. 状态管理与理赔金额聚合的踩坑记录

4.1 理赔金额聚合的状态后端选型

理赔场景需要按保单维度聚合一段时间内的理赔金额，比如「同一保单 30 天内累计理赔超过 10 万触发人工审核」。这个需求用 Flink 的KeyedProcessFunction配合ValueState实现，状态后端的选择直接影响性能和稳定性。

RocksDB 的配置有几个关键参数：state.backend.rocksdb.block.cache-size建议设为可用内存的 1/4，state.backend.rocksdb.writebuffer.size设为 64MB，state.backend.rocksdb.compaction.style用LEVEL而不是UNIVERSAL（理赔聚合的写放大不严重，LEVEL 更稳定）。

// 理赔金额聚合逻辑 public class ClaimAggregator extends KeyedProcessFunction<String, ClaimEvent, ClaimAlert> { private ValueState<BigDecimal> totalClaimState; private ValueState<Long> lastClaimTimeState; @Override public void open(Configuration parameters) { ValueStateDescriptor<BigDecimal> totalDesc = new ValueStateDescriptor<>("total-claim", TypeInformation.of(BigDecimal.class)); totalClaimState = getRuntimeContext().getState(totalDesc); ValueStateDescriptor<Long> timeDesc = new ValueStateDescriptor<>("last-claim-time", Types.LONG); lastClaimTimeState = getRuntimeContext().getState(timeDesc); } @Override public void processElement(ClaimEvent event, Context ctx, Collector<ClaimAlert> out) throws Exception { BigDecimal current = totalClaimState.value(); if (current == null) { current = BigDecimal.ZERO; } BigDecimal newTotal = current.add(event.getClaimAmount()); totalClaimState.update(newTotal); lastClaimTimeState.update(ctx.timestamp()); // 注册 30 天后的定时器，用于清理过期状态 ctx.timerService().registerEventTimeTimer(ctx.timestamp() + 30 * 24 * 60 * 60 * 1000L); if (newTotal.compareTo(new BigDecimal("100000")) > 0) { out.collect(new ClaimAlert(event.getPolicyNo(), newTotal, "EXCEED_THRESHOLD")); } } @Override public void onTimer(long timestamp, OnTimerContext ctx, Collector<ClaimAlert> out) { // 30 天窗口结束后清理状态，避免状态无限增长 totalClaimState.clear(); lastClaimTimeState.clear(); } }

registerEventTimeTimer注册的定时器在事件时间到达时触发onTimer，这里用来清理超过 30 天的状态。如果不做清理，状态会无限增长，最终导致 RocksDB 磁盘爆满。注意定时器的时间戳是基于事件时间的，如果水位线不推进，定时器永远不会触发——这也是为什么前面强调要设置withIdleness。

4.2 理赔聚合的常见问题排查

现象一：Checkpoint 持续失败，报错Checkpoint expired before completing。原因通常是状态太大导致 Checkpoint 时间超过超时阈值。解决方法是启用增量 Checkpoint（state.backend.incremental: true），同时把execution.checkpointing.timeout从默认的 10 分钟调到 30 分钟。如果还是失败，检查 RocksDB 的writebuffer是否频繁 flush，适当增大writebuffer.size。

现象二：作业运行几天后 TM 内存溢出。理赔聚合场景常见原因是状态没有清理逻辑，或者onTimer没有正确注册。排查方法是打开 Flink Web UI 的 Checkpoint 详情，看 State Size 是否持续增长。另一个可能原因是 RocksDB 的 Block Cache 和 Write Buffer 加起来超过了容器内存限制，需要调整taskmanager.memory.process.size并相应调低 RocksDB 内存参数。

现象三：理赔金额出现重复计算。如果 Kafka Source 的startingOffsets设为EARLIEST且没有开启 Checkpoint，作业重启后会从最早的数据重新消费。解决方法是确保 Checkpoint 开启且startingOffsets设为COMMITTED，同时下游 Sink 支持幂等写入（比如用保单号+理赔单号作为唯一键做 upsert）。

5. 生产部署与监控：从火焰图到血缘追踪

5.1 用火焰图定位反压与热点算子

Flink 作业上线后最常见的性能问题是反压（Backpressure）。Web UI 的反压指标只能告诉你哪个算子被压住了，但无法定位到具体代码行。这时候需要生成火焰图。Flink 内置了火焰图功能，在 JobManager Web UI 的「Flame Graph」页面可以按算子采样。

# 通过 REST API 触发火焰图采样（默认 30 秒） curl -X POST http://jobmanager:8081/jobs/<job-id>/vertices/<vertex-id>/flamegraph # 下载生成的火焰图 JSON 并转换为 SVG # 实际生产中更常用的是持续采样：每 5 分钟自动生成一次

火焰图里如果看到某个processElement方法占用大量 CPU 时间，通常是序列化/反序列化开销或者正则表达式匹配。保险核保规则里经常用正则做字段校验，比如身份证号、手机号格式验证，这些正则如果没预编译，每次调用都会重新编译 Pattern，CPU 消耗极大。解决方法是在open方法里预编译 Pattern 并缓存为成员变量。

另一个常见热点是KeyedProcessFunction里的ValueState读写。RocksDB 的读写虽然比堆内存慢，但正常情况下不会成为瓶颈。如果火焰图显示大量时间花在RocksDB.get上，检查是否在processElement里做了多次value()调用——每次调用都是一次 RocksDB 读取，应该一次性读出后缓存到局部变量。

5.2 用 OpenMetadata 追踪 Flink 作业的数据血缘

保险行业的数据治理要求能追溯每一条数据的来源和去向。Flink 作业的血缘关系包括：Source 表 → 中间算子 → Sink 表。OpenMetadata 支持通过 Flink 的 REST API 采集作业拓扑，自动生成血缘图。

# OpenMetadata Flink 采集配置 source: type: flink serviceName: insurance-flink-cluster serviceConnection: config: type: Flink hostPort: http://jobmanager:8081 env: prod sourceConfig: config: type: PipelineMetadata lineageInformation: dbServiceNames: - insurance-oracle - insurance-kafka sink: type: metadata-rest config: apiEndpoint: http://openmetadata-server:8585/api

配置中的dbServiceNames需要提前在 OpenMetadata 里注册好 Oracle 和 Kafka 的数据源。采集后，OpenMetadata 会解析 Flink 作业的Source和Sink定义，自动建立表级别的血缘关系。注意，如果 Flink 作业用的是自定义SourceFunction而不是 SQL/Table API，血缘解析可能不完整，需要手动补充。

血缘追踪的实际价值在于影响分析：当某个 Oracle 表要做结构变更时，可以通过血缘图快速找到所有依赖它的 Flink 作业，评估变更影响范围。我一般会在变更前跑一次血缘查询，把受影响的作业列表发给相关开发确认。

5.3 一个具体的调优技巧：Checkpoint 对齐与非对齐

Flink 的 Checkpoint 默认是对齐的（Aligned），即所有上游算子收到 barrier 后才开始快照。在反压严重时，对齐 Checkpoint 可能长时间无法完成。Flink 1.11 之后引入了非对齐 Checkpoint（Unaligned Checkpoint），允许 barrier 越过缓冲区中的数据继续向下游传递。

// 启用非对齐 Checkpoint env.getCheckpointConfig().enableUnalignedCheckpoints(); env.getCheckpointConfig().setAlignedCheckpointTimeout(Duration.ofSeconds(30));

setAlignedCheckpointTimeout表示先尝试对齐 Checkpoint，超过 30 秒后自动切换为非对齐。非对齐 Checkpoint 的代价是 Checkpoint 大小会增大（需要持久化缓冲区中的数据），所以只在反压严重且对齐 Checkpoint 持续超时时启用。保险理赔聚合作业在业务高峰期（比如每月 25 号后的理赔集中期）反压明显，我一般会开启这个配置，平时则关闭以减少 Checkpoint 存储开销。

最后说一个我自己的习惯：每次 Flink 作业上线前，先在预发环境用生产数据跑 24 小时，重点观察 Checkpoint 成功率、反压指标和状态大小增长曲线。这三个指标如果有任何一个不达标，坚决不上线。保险场景里，一个金额算错的 bug 可能需要几天才能发现，而修复成本远高于预发多跑一天。希望帮到你。

本文还有配套的精品资源，点击获取