简介这是一份面向大数据开发初学者与进阶工程师的Flink实战项目资料基于保险行业真实业务场景采用FlinkHBaseKafkaPhoenix架构实现业务系统数据库数据的实时同步与实时统计报表分析适合想通过完整项目理解流式计算落地流程的读者。资源包共73个文件约584KB以34个Java源码为核心辅以11个Python脚本、9个Shell脚本以及csv、properties、xml、sql、jar等配置与数据文件并附有说明文档覆盖从数据采集、处理到存储查询的完整链路。目前已有1166人学习下载。通过该资料读者可掌握实时数仓的模块划分与代码组织方式理解Kafka接入、Flink计算、HBase与Phoenix存储查询的协作逻辑并借助脚本与配置快速搭建本地运行环境同时作者提供Flink答疑服务便于快速入门与排查问题。1. 保险核心系统实时化为什么 Flink 成了那个绕不开的选项保险行业的核心系统有个特点白天跑批、晚上对账、月底出报表。这个节奏在过去二十年没什么问题直到业务方开始要求「保全申请提交后 3 秒内看到核保结论」「理赔报案后实时判断是否触发反欺诈规则」「渠道佣金按小时结算」。这些需求落到技术侧本质是把原来 T1 的批量计算压缩到秒级甚至毫秒级而且数据源横跨 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 格式 KafkaSourcePolicyChangeEvent source KafkaSource.PolicyChangeEventbuilder() .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(); DataStreamPolicyChangeEvent stream env.fromSource( source, WatermarkStrategy.PolicyChangeEventforBoundedOutOfOrderness(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 - 规则内容的映射 MapStateDescriptorString, UnderwritingRule ruleStateDesc new MapStateDescriptor( underwriting-rules, BasicTypeInfo.STRING_TYPE_INFO, TypeInformation.of(UnderwritingRule.class) ); // 规则流从 Kafka 或配置中心读取规则变更 DataStreamUnderwritingRule ruleStream env .addSource(new FlinkKafkaConsumer(underwriting-rules, new RuleDeserializer(), props)) .broadcast(ruleStateDesc); // 主流保单变更事件 DataStreamPolicyChangeEvent policyStream env .addSource(new FlinkKafkaConsumer(insurance-cdc.POLICY.POLICY_2024, new DebeziumJsonDeserializer(), props)) .keyBy(PolicyChangeEvent::getPolicyNo); // 连接广播流 BroadcastConnectedStreamPolicyChangeEvent, UnderwritingRule connected policyStream.connect(ruleStream); connected.process(new BroadcastProcessFunctionPolicyChangeEvent, UnderwritingRule, UnderwritingResult() { Override public void processElement(PolicyChangeEvent event, ReadOnlyContext ctx, CollectorUnderwritingResult out) throws Exception { // 从广播状态中读取当前生效的规则 MapStateString, 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, CollectorUnderwritingResult 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 配置 KafkaSinkUnderwritingResult sink KafkaSink.UnderwritingResultbuilder() .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设为 90000015 分钟是因为保险核保作业的 checkpoint 间隔通常设为 5 分钟事务超时时间需要大于 checkpoint 间隔的两倍以上否则事务可能在 checkpoint 完成前超时回滚。setTransactionalIdPrefix必须保证全局唯一否则多个作业实例会互相干扰。对账逻辑建议单独做一个离线校验作业每天凌晨读取前一天 Kafka 中的核保结果和 MySQL 中的核保记录按保单号做全量比对发现不一致时告警。这个校验作业不需要 Flink用 Spark 或直接 SQL 都能做关键是形成闭环。4. 状态管理与理赔金额聚合的踩坑记录4.1 理赔金额聚合的状态后端选型理赔场景需要按保单维度聚合一段时间内的理赔金额比如「同一保单 30 天内累计理赔超过 10 万触发人工审核」。这个需求用 Flink 的KeyedProcessFunction配合ValueState实现状态后端的选择直接影响性能和稳定性。状态后端适用场景保险理赔场景建议HashMapStateBackend状态小、对延迟敏感不推荐状态超过 1GB 后 GC 压力大EmbeddedRocksDBStateBackend状态大、读写频繁推荐理赔聚合状态通常几十 GB增量 Checkpoint状态大、Checkpoint 慢配合 RocksDB 使用减少 Checkpoint 时间RocksDB 的配置有几个关键参数state.backend.rocksdb.block.cache-size建议设为可用内存的 1/4state.backend.rocksdb.writebuffer.size设为 64MBstate.backend.rocksdb.compaction.style用LEVEL而不是UNIVERSAL理赔聚合的写放大不严重LEVEL 更稳定。// 理赔金额聚合逻辑 public class ClaimAggregator extends KeyedProcessFunctionString, ClaimEvent, ClaimAlert { private ValueStateBigDecimal totalClaimState; private ValueStateLong lastClaimTimeState; Override public void open(Configuration parameters) { ValueStateDescriptorBigDecimal totalDesc new ValueStateDescriptor(total-claim, TypeInformation.of(BigDecimal.class)); totalClaimState getRuntimeContext().getState(totalDesc); ValueStateDescriptorLong timeDesc new ValueStateDescriptor(last-claim-time, Types.LONG); lastClaimTimeState getRuntimeContext().getState(timeDesc); } Override public void processElement(ClaimEvent event, Context ctx, CollectorClaimAlert 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, CollectorClaimAlert out) { // 30 天窗口结束后清理状态避免状态无限增长 totalClaimState.clear(); lastClaimTimeState.clear(); } }registerEventTimeTimer注册的定时器在事件时间到达时触发onTimer这里用来清理超过 30 天的状态。如果不做清理状态会无限增长最终导致 RocksDB 磁盘爆满。注意定时器的时间戳是基于事件时间的如果水位线不推进定时器永远不会触发——这也是为什么前面强调要设置withIdleness。4.2 理赔聚合的常见问题排查现象一Checkpoint 持续失败报错Checkpoint expired before completing。原因通常是状态太大导致 Checkpoint 时间超过超时阈值。解决方法是启用增量 Checkpointstate.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 时间通常是序列化/反序列化开销或者正则表达式匹配。保险核保规则里经常用正则做字段校验比如身份证号、手机号格式验证这些正则如果没预编译每次调用都会重新编译 PatternCPU 消耗极大。解决方法是在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 之后引入了非对齐 CheckpointUnaligned Checkpoint允许 barrier 越过缓冲区中的数据继续向下游传递。// 启用非对齐 Checkpoint env.getCheckpointConfig().enableUnalignedCheckpoints(); env.getCheckpointConfig().setAlignedCheckpointTimeout(Duration.ofSeconds(30));setAlignedCheckpointTimeout表示先尝试对齐 Checkpoint超过 30 秒后自动切换为非对齐。非对齐 Checkpoint 的代价是 Checkpoint 大小会增大需要持久化缓冲区中的数据所以只在反压严重且对齐 Checkpoint 持续超时时启用。保险理赔聚合作业在业务高峰期比如每月 25 号后的理赔集中期反压明显我一般会开启这个配置平时则关闭以减少 Checkpoint 存储开销。最后说一个我自己的习惯每次 Flink 作业上线前先在预发环境用生产数据跑 24 小时重点观察 Checkpoint 成功率、反压指标和状态大小增长曲线。这三个指标如果有任何一个不达标坚决不上线。保险场景里一个金额算错的 bug 可能需要几天才能发现而修复成本远高于预发多跑一天。希望帮到你。本文还有配套的精品资源点击获取