简介面向大数据开发与Spark学习者的真实业务实战资源以保险行业实时数据分析为背景整合Kafka、Spark与HBase三大组件构建“数据采集—实时计算—结果存储—报表展示”的完整链路可帮助理解企业级实时数仓的典型架构。资源包共400个文件以28个Scala源码为主线辅以Shell脚本运行部署、properties与XML配置环境参数、JSON及sample样例数据业务模拟整体压缩后约632KB体量精简。目前已有979人学习下载。学习者可掌握Kafka Topic消费与Spark Streaming对接方式、窗口聚合与临时报表生成、结果写入HBase的持久化方案同时可参考保险保费收入、理赔监控、客户行为等业务指标的实时统计口径与计算思路适合希望结合真实场景提升Spark工程落地能力的初中级开发者也可用于教学实训或二次开发。1. 保险行业的实时报表为什么偏偏选 Spark 这套组合保险公司的数仓工程师被业务方追问“今天的保费收了多少”是常态麻烦的是上午问的是今天的下午问的就变成这个小时的了。离线跑批 T1 根本接不住这个节奏实时统计就成了刚需。这套《Spark实战项目保险行业真实项目》就是干这个的Kafka 承接业务系统数据库的变更事件Spark 做实时计算和窗口聚合HBase 存结果最终输出分钟级刷新的保费、理赔报表是一个真实业务形态的 Spark 数据分析案例不是 wordcount 那种玩具 demo。适合正在学 Spark 大数据处理、想找一个能照抄改成自己业务的人也适合准备搭实时数仓但还没定架构的工程师。2. 先把架构立住Kafka、Spark、HBase 各扛什么活2.1 三件套的分工与选型理由这套项目的链路看起来是三个组件拼在一起实际每个组件的位置都很讲究少一个或者换一个整个链路都会变味。Kafka 是数据管道。Apache Kafka 是开源的流处理平台在这个项目里扮演数据生产者和消费者之间的桥梁负责收集业务系统的数据库变更事件保证数据实时同步。保险业务系统每天产生几百万笔保单状态变更、理赔流转记录这些事件必须有个高吞吐的缓冲区Kafka 的 partition 机制天然支持并发写入和顺序消费加上副本容错broker 挂一台不影响数据完整性。生产环境常见做法是 replication.factor 至少设 2topic 按业务域拆比如 policy-events 和 claim-events 分开而不是全塞进一个 topic这样下游消费和排查都清爽。Spark 是计算引擎。Apache Spark 提供统一的编程模型批处理、交互式查询、流处理、机器学习都能跑。在这个项目里它从 Kafka 消费数据执行实时聚合、过滤、窗口操作生成报表指标。选 Spark 而不是纯写 Kafka Streams是因为报表逻辑里除了流计算还要跟历史维度做关联Spark SQL 处理这类“流 维表”的活更顺手分析师也能直接写 SQL 参与报表口径调整。Spark 的内存计算机制是提速核心同样的聚合逻辑在 MapReduce 上要跑几分钟的在 Spark 里秒级出结果。HBase 是结果仓库。Apache HBase 是基于 Hadoop 的分布式列式数据库适合存大规模、稀疏的数据。Spark 算完的结果持久化到 HBase报表服务直接查。保险数据的稀疏性很典型——车险保单和健康险保单字段差异巨大用列式存储不浪费空间HBase 的强一致性和水平扩展能力让它比直接写 MySQL 或 Redis 更能扛住报表查询压力。有人会问现在 Flink 那么火为什么不直接 Flink ClickHouse这个问题我在好几个项目里都被问过答案就一句话选型第一原则是团队能不能维护而不是技术够不够新。很多保险公司的数据部门已经有一大批 Spark 离线任务Spark 的 RDD、DataFrame、SQL 心智模型大家都熟复用同一套技能栈做实时出问题能接得住。ClickHouse 做报表确实快但等于再养一套独立集群。这套资源的定位是让你在一套熟悉的体系里跑通实时链路而不是堆一堆新技术。真到生产环境如果团队已经很熟 Flink把 Spark 消费逻辑平移过去是可以的但那是后话。2.2 一条完整数据流五个环节各司其职整个项目实现流程按五个环节走每个环节的产出和输入都很清晰环节核心组件产出数据生产Kafka Producer / CanalKafka topic 里的 JSON 事件数据消费Spark Streaming解析后的 DataFrame数据计算Spark 窗口聚合分钟级统计指标数据存储HBaseinsurance_report 表实时报表查询接口 前端分钟级刷新的报表这个流程里最有价值的不是每一步单独的技术而是“生产到报表”的完整闭环。很多学习项目只教你 Spark 消费 Kafka 然后打印日志或者只教你用 HBase 存一个 wordcount 结果这套项目是真正把保险业务场景串起来了。数据生产环节里业务系统产生的数据库变更事件被写入 Kafka topic数据消费环节Spark Streaming 从 Kafka topic 里读出来数据计算环节做实时聚合和窗口操作数据存储环节落到 HBase最后实时报表接口对接 HBase 提供查询。后面第 3、4 章就是照这个顺序往下拆的。2.3 环境搭建与参数选型spark集群搭建不是最难的配参数才是拿到这套资源后第一步自然是把环境跑起来。spark集群搭建的教程网上很多三台节点、一台 master 两台 worker 是最常见的起步配置。我提醒一句环境搭建不是这个项目的重点参数选型才是。同样一套代码参数不对跑起来要么 OOM 要么慢得像蜗牛。参数建议值说明Kafka replication.factor2至少 2broker 挂一台不丢数据topic 分区数executor 总 core 的 2~3 倍分区上限决定消费并行度spark.executor.memory4G~8G看单条事件大小和窗口聚合状态量spark.streaming.kafka.maxRatePerPartition1000~2000限速防止积压数据一次性灌进内存hbase.hregion.memstore.flush.size128M默认值写放大明显时再调checkpointLocation独立 HDFS 目录每个环境一套别共用分区数这块很多人栽跟头。topic 分区数定小了executor 再多也跑不满定大了Kafka 和 HBase 的 IO 压力全上来。常见做法是先按 executor 总 core 数的 2 倍设跑起来看 Spark UI 的 batch 处理时间如果处理时间明显小于 batch 间隔说明资源有富余可以再压分区数或者缩集群。版本兼容性也得提前确认。Spark 消费 Kafka 用的是 spark-sql-kafka-0-10 连接器它要求 Kafka broker 版本在 0.10 以上这个门槛现在基本都满足。但要注意Structured Streaming 读 Kafka 时不支持 Kafka 3.x 引入的一些新 consumer 参数配置时按 0.10 的语义来写最稳别在 option 里塞高版本才有的参数否则启动直接报错而且报错信息很隐晦看起来像网络问题实际是参数不识别。3. 数据生产端把业务系统的数据库变更写进 Kafka3.1 模拟事件生成保单和理赔数据长什么样真实保险项目里业务系统的数据库变更事件常见做法是用 Canal 监听 MySQL binlog解析成结构化事件后发到 Kafka。这套资源为了让你在一台笔记本上就能复现生产端直接用 Python 模拟程序生成保单和理赔事件逻辑和 Canal 产出的等价——都是把业务动作变成一条带事件类型的 JSON。import json import random import time from datetime import datetime def gen_policy_event(): return { event_type: policy, event_id: fP{int(time.time() * 1000)}{random.randint(1000, 9999)}, policy_id: fP{random.randint(10000000, 99999999)}, user_id: fU{random.randint(100000, 999999)}, product_type: random.choice([auto, health, accident]), premium: round(random.uniform(800, 12000), 2), event_time: datetime.now().strftime(%Y-%m-%d %H:%M:%S) } def gen_claim_event(): return { event_type: claim, event_id: fC{int(time.time() * 1000)}{random.randint(1000, 9999)}, claim_id: fC{random.randint(10000000, 99999999)}, policy_id: fP{random.randint(10000000, 99999999)}, claim_amount: round(random.uniform(2000, 50000), 2), status: random.choice([submitted, approved, paid]), event_time: datetime.now().strftime(%Y-%m-%d %H:%M:%S) }字段说明policy 事件带 product_type 和 premium车险 auto、健康险 health、意外险 accident 三种产品分开统计claim 事件带 claim_amount 和 status。event_time 统一用字符串格式这个格式一致性很重要后面 Spark 解析时间戳时如果格式不统一from_json 会直接解析失败返回 null。event_id 是全局唯一的事件 ID这是为下游幂等去重留的口子第 5 章会专门讲为什么必须有它。3.2 Producer 参数配置acks、batch.size、linger.ms 怎么定Kafka Producer 的配置直接影响数据生产端的可靠性。默认配置能用但保险场景下丢一条数据报表数字就对不上所以核心参数都得显式设from kafka import KafkaProducer producer KafkaProducer( bootstrap_serversnode01:9092,node02:9092,node03:9092, acksall, retries3, batch_size16384, linger_ms100, compression_typelz4, value_serializerlambda v: json.dumps(v, ensure_asciiFalse).encode(utf-8) ) while True: if random.random() 0.7: event gen_policy_event() else: event gen_claim_event() producer.send(insurance-events, valueevent) time.sleep(0.05)参数值为什么acksall所有 ISR 副本确认才算成功避免 leader 写成功就返回导致丢数据retries3网络抖动时自动重发但不解决重复问题下游要靠 event_id 去重batch.size1638416KB攒够一批再发减少小消息的网络开销linger.ms100最多等 100ms 凑批兼顾延迟和吞吐compression_typelz4压缩省带宽保险事件 JSON 冗余度高压缩比很可观这里有个细节真实业务系统的变更事件量级是波动的白天高峰、凌晨低谷。如果用固定 sleep 模拟Spark 那边看到的流量是匀速的起不到压测效果。我一般把 sleep 改成随机值比如time.sleep(random.uniform(0.01, 0.2))模拟出波峰波谷后面调试限速参数才有意义。生产端跑起来之后先别急着起 Spark用下面小节的方法确认数据真的进 topic 了再往下走。3.3 先验证数据进没进 Kafkaconsole consumer 看两眼kafka-console-consumer.sh \ --bootstrap-server node01:9092,node02:9092,node03:9092 \ --topic insurance-events \ --from-beginning \ --max-messages 5看到 5 条 JSON 事件就说明生产端通了。注意--from-beginning是从头开始消费验证环境无所谓生产环境排查问题时慎用会把积压数据全拉出来。确认完内容之后再看消费积压情况kafka-run-class.sh kafka.tools.GetOffsetShell \ --bootstrap-server node01:9092 \ --topic insurance-events \ --time -1这个命令打印每个分区的 latest offset和当前消费组的 committed offset 一减就是积压量。实时项目里积压量是最重要的健康指标。Spark 消费端还没起的时候你会看到积压持续上涨这是正常的说明数据在进 Kafka等 Spark 起来后积压应该逐步回落。如果积压一直涨大概率是消费端处理速度跟不上回到第 2 章那个参数表去调。4. Spark 消费与实时计算从 Kafka 到 HBase 的统计结果4.1 消费配置偏移量管理和限速参数Spark 消费 Kafka 这一段新手最容易踩的坑是把所有参数都按默认来。Structured Streaming 读 Kafka 时有几个关键 option 必须显式指定否则要么丢数据要么 OOM。import org.apache.spark.sql.SparkSession val spark SparkSession.builder() .appName(insurance-realtime-report) .config(spark.streaming.kafka.maxRatePerPartition, 1000) .getOrCreate() val rawDF spark.readStream .format(kafka) .option(kafka.bootstrap.servers, node01:9092,node02:9092,node03:9092) .option(subscribe, insurance-events) .option(startingOffsets, earliest) .option(enable.auto.commit, false) .load()几个参数逐个说。kafka.bootstrap.servers指向 broker 地址跟生产端一致。subscribe指定 topicStructured Streaming 支持逗号分隔多 topic。startingOffsets是首次启动时的消费位置earliest 表示从最早开始适合报表这种要补全量的场景只要增量就设 latest。enable.auto.commit要设成 false这是血泪经验——Structured Streaming 靠 checkpoint 管理偏移如果还开着 Kafka 的自动提交两边各管各的任务一旦重启要么重复消费要么丢数据完全不可控。maxRatePerPartition是限速开关默认不限制。保险业务高峰期 Kafka 里积压几十万条事件是常态任务刚启动时如果一口气全拉进内存executor 直接 OOM。设成 1000 意味着每个分区每秒最多消费 1000 条4 个分区就是每秒 4000 条足够报表用又不会打爆内存。这个值别拍脑袋定先设 500 跑一个 batch 看处理时间再逐步往上调。4.2 窗口聚合按产品类型统计保费和理赔从 Kafka 读出来的是二进制的 value 字段第一步得转成字符串再用 from_json 按 schema 解析——这就是 Spark 中读取 JSON 的标准姿势很多 spark 数据分析案例也是这一步开始分叉的。import org.apache.spark.sql.functions._ import org.apache.spark.sql.types._ val schema StructType(Seq( StructField(event_type, StringType, true), StructField(event_id, StringType, true), StructField(policy_id, StringType, true), StructField(user_id, StringType, true), StructField(product_type, StringType, true), StructField(premium, DoubleType, true), StructField(claim_amount, DoubleType, true), StructField(status, StringType, true), StructField(event_time, TimestampType, true) )) val parsedDF rawDF .selectExpr(CAST(value AS STRING) AS json_str) .select(from_json($json_str, schema).alias(data)) .select(data.*) val reportDF parsedDF .withWatermark(event_time, 60 seconds) .groupBy( window($event_time, 5 minutes, 1 minute), $product_type ) .agg( sum(when($event_type policy, $premium).otherwise(0)).alias(premium_total), count(when($event_type claim, 1)).alias(claim_count), sum(when($event_type claim, $claim_amount).otherwise(0)).alias(claim_amount_total) )逻辑说明from_json 需要先给一个 schema 定义Spark 按这个 schema 解析 JSON 字符串字段对不上会返回 null——所以 schema 里每个字段都标了 nullable true宁可字段是 null 也不能让整条解析失败。窗口聚合用window(event_time, 5 minutes, 1 minute)表示 5 分钟窗口、1 分钟滑动一次也就是说每分钟都会输出最近 5 分钟的保费和理赔统计报表做到分钟级刷新靠的就是这个参数。watermark 设 60 秒允许事件时间比处理时间晚最多 60 秒晚于阈值的数据直接丢弃避免旧数据反复触发窗口重算。报表口径上premium_total 只在 event_type 为 policy 时累加claim 事件记入 claim_count 和 claim_amount_total。这里用when().otherwise()而不是先 filter 再聚合是因为窗口的语义要求所有事件都留在分组里提前过滤会导致窗口时间段内的事件维度缺失统计口径就错了。4.3 结果写 HBaseRowKey 设计决定查询性能结果写 HBase 是整个链路里影响后续查询性能最关键的一步。RowKey 设计不好写入热点、查询慢、region 频繁分裂报表接口全卡。先看写入代码的骨架import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Put} import org.apache.hadoop.hbase.util.Bytes import org.apache.hadoop.hbase.{HBaseConfiguration, TableName} import org.apache.spark.sql.streaming.Trigger def buildRowKey(product: String, windowStart: Long, bucket: Int): String { val reverseTs (Long.MaxValue - windowStart).toString f$bucket%02d_${product}_$reverseTs } reportDF.writeStream .foreachBatch { (batchDF, batchId) batchDF.foreachPartition { rows val conn ConnectionFactory.createConnection(HBaseConfiguration.create()) val table conn.getTable(TableName.valueOf(insurance_report)) rows.foreach { row val win row.getAs[org.apache.spark.sql.Row](window) val windowStartTs win.getTimestamp(0).getTime val rowKey buildRowKey( row.getAs[String](product_type), windowStartTs, (windowStartTs % 97).toInt ) val put new Put(Bytes.toBytes(rowKey)) put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(premium_total), Bytes.toBytes(row.getAs[Double](premium_total))) put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(claim_count), Bytes.toBytes(row.getAs[Long](claim_count))) put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(claim_amount_total), Bytes.toBytes(row.getAs[Double](claim_amount_total))) table.put(put) } table.close() conn.close() } } .trigger(Trigger.ProcessingTime(60 seconds)) .option(checkpointLocation, /data/checkpoint/insurance_report) .start()RowKey 的构造是这段代码的重点。如果把时间戳直接放 RowKey 开头同一分钟写入的数据会顺序落在同一个 region 上形成写热点——这是 HBase 最常见的翻车姿势。这里做了两件事第一bucket 取时间戳对 97 取模把同一分钟的数据打散到 97 个分桶写入压力分散到不同 region第二时间戳用 Long.MaxValue 减去原值再转字符串让新的时间戳排前面按时间倒序扫描时 HBase 的 range scan 效率最高。生产环境里桶数按 region 数量调一般是 region 数的整数倍。表结构就一个列族 cf列分别存 premium_total、claim_count、claim_amount_total。列族别建多了HBase 一个列族对应一个 MemStore列族太多内存开销大报表场景一个列族足够。trigger设 ProcessingTime 60 秒控制结果写 HBase 的节奏跟下游报表刷新频率对齐。5. 避坑指南保险实时项目最容易翻车的六个点这套链路我前后跑过三遍才跑顺下面这些坑每个都真实发生过。按现象、原因、解决三步写清楚你照着排查就行。5.1 Executor 频繁 OOM任务无限重启现象Spark UI 里 executor 一个接一个挂掉日志报 java heap space任务重启后过一会儿又挂形成死循环。原因Kafka 里积压了大量事件maxRatePerPartition 没设或者设得太大executor 一次性拉了几十万条进内存窗口聚合的中间状态也全堆在堆内spark 内存分配根本扛不住。解决给每个分区限速spark.streaming.kafka.maxRatePerPartition从 500 开始调窗口聚合的状态用独立的 state store 目录别和 shuffle 目录挤在一起。血泪经验先限速再调内存顺序别搞反。一上来就加 executor memory治标不治本积压还在迟早再炸。5.2 报表数字和业务系统对不上现象报表显示今天保费收入 1200 万业务系统自己的统计是 1250 万差值看起来没规律今天多明天少。原因Kafka 生产端重试导致事件重复加上 Spark 任务重启后从 checkpoint 恢复时部分 batch 重复消费而结果写入 HBase 不是幂等的同一笔保费被累加了两次。解决三层防护。第一层生产端每条事件带全局唯一 event_id第二层HBase 写入前用 event_id 做去重或者把累加逻辑改成 HBase 的 increment 操作而不是 put 覆盖第三层checkpoint 用独立目录别让两个环境共用。这三层缺一层数字都对不齐。5.3 HBase 写热点一个 region 被打满现象HBase 监控里某个 region 的写请求数是其他 region 的几十倍region 频繁分裂写入延迟飙高报表查接口跟着变慢。原因RowKey 用了时间戳开头或者产品类型开头。时间戳顺序写必然打到一个 region产品类型只有三种热点更明显。这是 RowKey 设计问题不是 HBase 本身的问题。解决RowKey 加盐分桶时间戳反转把热门 key 打散第 4 章 buildRowKey 里已经给了实现。桶数别拍脑袋定按 region 数来上线后观察写入分布再调。加了盐之后 range scan 可能受影响所以加盐字段要放在 RowKey 低位时间戳反转放高位保证按时间查询还是高效的。5.4 窗口统计结果偏少数据“丢了”现象报表统计的理赔笔数明显少于业务系统但 Kafka 里数据一条没少Kafka 积压也是正常的。原因事件乱序。业务系统的事件到达 Kafka 的时间可能比事件产生时间晚几十秒甚至几分钟窗口按 event_time 聚合时晚到的数据落进了上一个窗口但那个窗口已经触发输出并关闭了没有 watermark 兜底数据就被丢了。解决withWatermark 设置合理的延迟阈值保险场景 60 秒起步同时打开 allowedLateness让晚到数据能触发窗口重新计算输出更新后的结果。这里要区分watermark 决定数据还等不等allowedLateness 决定晚到数据来了之后要不要重算两个配合着用才算完整。5.5 JSON schema 一变整批数据全变 null现象某天开始报表统计指标全是 0或者某一列全是空日志里 from_json 解析看起来没报错但数据结果不对。原因业务系统给事件里加了字段或者把某个字段类型改了比如 premium 从数字变成字符串。from_json 按旧 schema 解析新数据字段对不上返回 null聚合结果自然全错。这个坑最阴险因为它不报错只是静默变 null。解决schema 演进要跟着业务走解析层加一层校验——统计 json_str 里字段解析失败的条数超过阈值就报警同时在 from_json 外面包一层 try 语义脏数据单独写到一个 dead-letter topic别让脏数据混进主链路。别小看这个保险业务系统上线后改字段是常态不加这层校验等于埋雷。5.6 本地跑得好好的集群上任务起不来现象代码本地 IDEA 里跑通过打成 jar 丢到集群上启动报 ClassNotFoundException或者 JSON 解析行为跟本地完全不一样。原因八成是依赖冲突。本地用的是新版本 Kafka client集群上 Spark 自带的 kafka-clients 版本旧运行时类加载用的是集群那份另外 Scala 版本不一致也会出这种问题。解决打 fat jar 时把 spark-sql-kafka-0-10 的传递依赖排除掉尤其是 kafka-clients用集群自带的版本Spark 用 2.4 就配 Scala 2.11 的包用 3.x 就配 Scala 2.12 或 2.13别混。部署前先在集群上跑一个最小消费任务验证环境再上完整代码别把排查时间浪费在环境问题上。6. 上线前的验证与监控从「能跑」到「能上线」的最后一公里代码跑通只是第一步真正上线前我一般会花半天时间做三件事对账、压延迟、盯监控。这三件事做完才敢说这套实时链路是可靠的。对账是验证数据准确性的硬办法。挑一个时间窗口把 Kafka 里该窗口的事件一条条数出来跟 HBase 里对应窗口的聚合结果做对比。保费总额可以从事件明细累加验证理赔笔数直接数 claim 事件条数。两边对不上就回到第 5 章的幂等和乱序问题逐项排查。这个步骤我每次上线都做不做的话报表上线第二天业务方准来问数字对不对。压延迟是验证实时性的关键。给事件打上生产时间戳在 HBase 写入列里记录处理时间戳两者差值就是端到端延迟。保险报表场景目标一般是分钟级如果发现延迟超过 3 分钟先看 Kafka 积压量再看 Spark UI 的 batch 处理时间是不是逼近 batch 间隔最后检查 HBase 写入是否有热点。指标怎么看阈值参考batch 处理时间Spark UI Structured Streaming 页低于 batch 间隔的 60%Kafka 积压GetOffsetShell 的 latest 减 committed持续增长才报警端到端延迟处理时间戳减事件时间戳目标分钟级GC 频率jstat -gcutilFull GC 接近 0盯监控建议把 Spark UI 的 Structured Streaming 页面作为默认入口batch 处理时间、输入行数、watermark 水位这几个指标一眼就能看到瓶颈。内存和线程层面的问题用 spark 内存线程监测工具查——jstat 看老年代和 GC 频率jstack 看是不是有线程卡在 HBase 的 RPC 上。GC 频繁基本是状态太大或者限速不够线程卡 RPC 基本是 HBase 热点写或者连接池泄漏方向很明确。有一个习惯我一直保留着任何实时任务上线之前强制走一遍“从埋点到报表”的完整对账流程不管这个任务改了几个字段、动了哪个窗口参数。因为实时链路里每个环节都会不定时出幺蛾子Kafka 会重复投递Spark 会重启HBase 会分裂谁都不能保证稳定性只有对账能兜底。这套保险实时项目跑通之后我把同样的流程搬到了另外两个业务线把对账和监控脚本直接复用省掉了大量排查时间。希望帮到你。本文还有配套的精品资源点击获取
简介面向大数据开发与Spark学习者的真实业务实战资源以保险行业实时数据分析为背景整合Kafka、Spark与HBase三大组件构建“数据采集—实时计算—结果存储—报表展示”的完整链路可帮助理解企业级实时数仓的典型架构。资源包共400个文件以28个Scala源码为主线辅以Shell脚本运行部署、properties与XML配置环境参数、JSON及sample样例数据业务模拟整体压缩后约632KB体量精简。目前已有979人学习下载。学习者可掌握Kafka Topic消费与Spark Streaming对接方式、窗口聚合与临时报表生成、结果写入HBase的持久化方案同时可参考保险保费收入、理赔监控、客户行为等业务指标的实时统计口径与计算思路适合希望结合真实场景提升Spark工程落地能力的初中级开发者也可用于教学实训或二次开发。1. 保险行业的实时报表为什么偏偏选 Spark 这套组合保险公司的数仓工程师被业务方追问“今天的保费收了多少”是常态麻烦的是上午问的是今天的下午问的就变成这个小时的了。离线跑批 T1 根本接不住这个节奏实时统计就成了刚需。这套《Spark实战项目保险行业真实项目》就是干这个的Kafka 承接业务系统数据库的变更事件Spark 做实时计算和窗口聚合HBase 存结果最终输出分钟级刷新的保费、理赔报表是一个真实业务形态的 Spark 数据分析案例不是 wordcount 那种玩具 demo。适合正在学 Spark 大数据处理、想找一个能照抄改成自己业务的人也适合准备搭实时数仓但还没定架构的工程师。2. 先把架构立住Kafka、Spark、HBase 各扛什么活2.1 三件套的分工与选型理由这套项目的链路看起来是三个组件拼在一起实际每个组件的位置都很讲究少一个或者换一个整个链路都会变味。Kafka 是数据管道。Apache Kafka 是开源的流处理平台在这个项目里扮演数据生产者和消费者之间的桥梁负责收集业务系统的数据库变更事件保证数据实时同步。保险业务系统每天产生几百万笔保单状态变更、理赔流转记录这些事件必须有个高吞吐的缓冲区Kafka 的 partition 机制天然支持并发写入和顺序消费加上副本容错broker 挂一台不影响数据完整性。生产环境常见做法是 replication.factor 至少设 2topic 按业务域拆比如 policy-events 和 claim-events 分开而不是全塞进一个 topic这样下游消费和排查都清爽。Spark 是计算引擎。Apache Spark 提供统一的编程模型批处理、交互式查询、流处理、机器学习都能跑。在这个项目里它从 Kafka 消费数据执行实时聚合、过滤、窗口操作生成报表指标。选 Spark 而不是纯写 Kafka Streams是因为报表逻辑里除了流计算还要跟历史维度做关联Spark SQL 处理这类“流 维表”的活更顺手分析师也能直接写 SQL 参与报表口径调整。Spark 的内存计算机制是提速核心同样的聚合逻辑在 MapReduce 上要跑几分钟的在 Spark 里秒级出结果。HBase 是结果仓库。Apache HBase 是基于 Hadoop 的分布式列式数据库适合存大规模、稀疏的数据。Spark 算完的结果持久化到 HBase报表服务直接查。保险数据的稀疏性很典型——车险保单和健康险保单字段差异巨大用列式存储不浪费空间HBase 的强一致性和水平扩展能力让它比直接写 MySQL 或 Redis 更能扛住报表查询压力。有人会问现在 Flink 那么火为什么不直接 Flink ClickHouse这个问题我在好几个项目里都被问过答案就一句话选型第一原则是团队能不能维护而不是技术够不够新。很多保险公司的数据部门已经有一大批 Spark 离线任务Spark 的 RDD、DataFrame、SQL 心智模型大家都熟复用同一套技能栈做实时出问题能接得住。ClickHouse 做报表确实快但等于再养一套独立集群。这套资源的定位是让你在一套熟悉的体系里跑通实时链路而不是堆一堆新技术。真到生产环境如果团队已经很熟 Flink把 Spark 消费逻辑平移过去是可以的但那是后话。2.2 一条完整数据流五个环节各司其职整个项目实现流程按五个环节走每个环节的产出和输入都很清晰环节核心组件产出数据生产Kafka Producer / CanalKafka topic 里的 JSON 事件数据消费Spark Streaming解析后的 DataFrame数据计算Spark 窗口聚合分钟级统计指标数据存储HBaseinsurance_report 表实时报表查询接口 前端分钟级刷新的报表这个流程里最有价值的不是每一步单独的技术而是“生产到报表”的完整闭环。很多学习项目只教你 Spark 消费 Kafka 然后打印日志或者只教你用 HBase 存一个 wordcount 结果这套项目是真正把保险业务场景串起来了。数据生产环节里业务系统产生的数据库变更事件被写入 Kafka topic数据消费环节Spark Streaming 从 Kafka topic 里读出来数据计算环节做实时聚合和窗口操作数据存储环节落到 HBase最后实时报表接口对接 HBase 提供查询。后面第 3、4 章就是照这个顺序往下拆的。2.3 环境搭建与参数选型spark集群搭建不是最难的配参数才是拿到这套资源后第一步自然是把环境跑起来。spark集群搭建的教程网上很多三台节点、一台 master 两台 worker 是最常见的起步配置。我提醒一句环境搭建不是这个项目的重点参数选型才是。同样一套代码参数不对跑起来要么 OOM 要么慢得像蜗牛。参数建议值说明Kafka replication.factor2至少 2broker 挂一台不丢数据topic 分区数executor 总 core 的 2~3 倍分区上限决定消费并行度spark.executor.memory4G~8G看单条事件大小和窗口聚合状态量spark.streaming.kafka.maxRatePerPartition1000~2000限速防止积压数据一次性灌进内存hbase.hregion.memstore.flush.size128M默认值写放大明显时再调checkpointLocation独立 HDFS 目录每个环境一套别共用分区数这块很多人栽跟头。topic 分区数定小了executor 再多也跑不满定大了Kafka 和 HBase 的 IO 压力全上来。常见做法是先按 executor 总 core 数的 2 倍设跑起来看 Spark UI 的 batch 处理时间如果处理时间明显小于 batch 间隔说明资源有富余可以再压分区数或者缩集群。版本兼容性也得提前确认。Spark 消费 Kafka 用的是 spark-sql-kafka-0-10 连接器它要求 Kafka broker 版本在 0.10 以上这个门槛现在基本都满足。但要注意Structured Streaming 读 Kafka 时不支持 Kafka 3.x 引入的一些新 consumer 参数配置时按 0.10 的语义来写最稳别在 option 里塞高版本才有的参数否则启动直接报错而且报错信息很隐晦看起来像网络问题实际是参数不识别。3. 数据生产端把业务系统的数据库变更写进 Kafka3.1 模拟事件生成保单和理赔数据长什么样真实保险项目里业务系统的数据库变更事件常见做法是用 Canal 监听 MySQL binlog解析成结构化事件后发到 Kafka。这套资源为了让你在一台笔记本上就能复现生产端直接用 Python 模拟程序生成保单和理赔事件逻辑和 Canal 产出的等价——都是把业务动作变成一条带事件类型的 JSON。import json import random import time from datetime import datetime def gen_policy_event(): return { event_type: policy, event_id: fP{int(time.time() * 1000)}{random.randint(1000, 9999)}, policy_id: fP{random.randint(10000000, 99999999)}, user_id: fU{random.randint(100000, 999999)}, product_type: random.choice([auto, health, accident]), premium: round(random.uniform(800, 12000), 2), event_time: datetime.now().strftime(%Y-%m-%d %H:%M:%S) } def gen_claim_event(): return { event_type: claim, event_id: fC{int(time.time() * 1000)}{random.randint(1000, 9999)}, claim_id: fC{random.randint(10000000, 99999999)}, policy_id: fP{random.randint(10000000, 99999999)}, claim_amount: round(random.uniform(2000, 50000), 2), status: random.choice([submitted, approved, paid]), event_time: datetime.now().strftime(%Y-%m-%d %H:%M:%S) }字段说明policy 事件带 product_type 和 premium车险 auto、健康险 health、意外险 accident 三种产品分开统计claim 事件带 claim_amount 和 status。event_time 统一用字符串格式这个格式一致性很重要后面 Spark 解析时间戳时如果格式不统一from_json 会直接解析失败返回 null。event_id 是全局唯一的事件 ID这是为下游幂等去重留的口子第 5 章会专门讲为什么必须有它。3.2 Producer 参数配置acks、batch.size、linger.ms 怎么定Kafka Producer 的配置直接影响数据生产端的可靠性。默认配置能用但保险场景下丢一条数据报表数字就对不上所以核心参数都得显式设from kafka import KafkaProducer producer KafkaProducer( bootstrap_serversnode01:9092,node02:9092,node03:9092, acksall, retries3, batch_size16384, linger_ms100, compression_typelz4, value_serializerlambda v: json.dumps(v, ensure_asciiFalse).encode(utf-8) ) while True: if random.random() 0.7: event gen_policy_event() else: event gen_claim_event() producer.send(insurance-events, valueevent) time.sleep(0.05)参数值为什么acksall所有 ISR 副本确认才算成功避免 leader 写成功就返回导致丢数据retries3网络抖动时自动重发但不解决重复问题下游要靠 event_id 去重batch.size1638416KB攒够一批再发减少小消息的网络开销linger.ms100最多等 100ms 凑批兼顾延迟和吞吐compression_typelz4压缩省带宽保险事件 JSON 冗余度高压缩比很可观这里有个细节真实业务系统的变更事件量级是波动的白天高峰、凌晨低谷。如果用固定 sleep 模拟Spark 那边看到的流量是匀速的起不到压测效果。我一般把 sleep 改成随机值比如time.sleep(random.uniform(0.01, 0.2))模拟出波峰波谷后面调试限速参数才有意义。生产端跑起来之后先别急着起 Spark用下面小节的方法确认数据真的进 topic 了再往下走。3.3 先验证数据进没进 Kafkaconsole consumer 看两眼kafka-console-consumer.sh \ --bootstrap-server node01:9092,node02:9092,node03:9092 \ --topic insurance-events \ --from-beginning \ --max-messages 5看到 5 条 JSON 事件就说明生产端通了。注意--from-beginning是从头开始消费验证环境无所谓生产环境排查问题时慎用会把积压数据全拉出来。确认完内容之后再看消费积压情况kafka-run-class.sh kafka.tools.GetOffsetShell \ --bootstrap-server node01:9092 \ --topic insurance-events \ --time -1这个命令打印每个分区的 latest offset和当前消费组的 committed offset 一减就是积压量。实时项目里积压量是最重要的健康指标。Spark 消费端还没起的时候你会看到积压持续上涨这是正常的说明数据在进 Kafka等 Spark 起来后积压应该逐步回落。如果积压一直涨大概率是消费端处理速度跟不上回到第 2 章那个参数表去调。4. Spark 消费与实时计算从 Kafka 到 HBase 的统计结果4.1 消费配置偏移量管理和限速参数Spark 消费 Kafka 这一段新手最容易踩的坑是把所有参数都按默认来。Structured Streaming 读 Kafka 时有几个关键 option 必须显式指定否则要么丢数据要么 OOM。import org.apache.spark.sql.SparkSession val spark SparkSession.builder() .appName(insurance-realtime-report) .config(spark.streaming.kafka.maxRatePerPartition, 1000) .getOrCreate() val rawDF spark.readStream .format(kafka) .option(kafka.bootstrap.servers, node01:9092,node02:9092,node03:9092) .option(subscribe, insurance-events) .option(startingOffsets, earliest) .option(enable.auto.commit, false) .load()几个参数逐个说。kafka.bootstrap.servers指向 broker 地址跟生产端一致。subscribe指定 topicStructured Streaming 支持逗号分隔多 topic。startingOffsets是首次启动时的消费位置earliest 表示从最早开始适合报表这种要补全量的场景只要增量就设 latest。enable.auto.commit要设成 false这是血泪经验——Structured Streaming 靠 checkpoint 管理偏移如果还开着 Kafka 的自动提交两边各管各的任务一旦重启要么重复消费要么丢数据完全不可控。maxRatePerPartition是限速开关默认不限制。保险业务高峰期 Kafka 里积压几十万条事件是常态任务刚启动时如果一口气全拉进内存executor 直接 OOM。设成 1000 意味着每个分区每秒最多消费 1000 条4 个分区就是每秒 4000 条足够报表用又不会打爆内存。这个值别拍脑袋定先设 500 跑一个 batch 看处理时间再逐步往上调。4.2 窗口聚合按产品类型统计保费和理赔从 Kafka 读出来的是二进制的 value 字段第一步得转成字符串再用 from_json 按 schema 解析——这就是 Spark 中读取 JSON 的标准姿势很多 spark 数据分析案例也是这一步开始分叉的。import org.apache.spark.sql.functions._ import org.apache.spark.sql.types._ val schema StructType(Seq( StructField(event_type, StringType, true), StructField(event_id, StringType, true), StructField(policy_id, StringType, true), StructField(user_id, StringType, true), StructField(product_type, StringType, true), StructField(premium, DoubleType, true), StructField(claim_amount, DoubleType, true), StructField(status, StringType, true), StructField(event_time, TimestampType, true) )) val parsedDF rawDF .selectExpr(CAST(value AS STRING) AS json_str) .select(from_json($json_str, schema).alias(data)) .select(data.*) val reportDF parsedDF .withWatermark(event_time, 60 seconds) .groupBy( window($event_time, 5 minutes, 1 minute), $product_type ) .agg( sum(when($event_type policy, $premium).otherwise(0)).alias(premium_total), count(when($event_type claim, 1)).alias(claim_count), sum(when($event_type claim, $claim_amount).otherwise(0)).alias(claim_amount_total) )逻辑说明from_json 需要先给一个 schema 定义Spark 按这个 schema 解析 JSON 字符串字段对不上会返回 null——所以 schema 里每个字段都标了 nullable true宁可字段是 null 也不能让整条解析失败。窗口聚合用window(event_time, 5 minutes, 1 minute)表示 5 分钟窗口、1 分钟滑动一次也就是说每分钟都会输出最近 5 分钟的保费和理赔统计报表做到分钟级刷新靠的就是这个参数。watermark 设 60 秒允许事件时间比处理时间晚最多 60 秒晚于阈值的数据直接丢弃避免旧数据反复触发窗口重算。报表口径上premium_total 只在 event_type 为 policy 时累加claim 事件记入 claim_count 和 claim_amount_total。这里用when().otherwise()而不是先 filter 再聚合是因为窗口的语义要求所有事件都留在分组里提前过滤会导致窗口时间段内的事件维度缺失统计口径就错了。4.3 结果写 HBaseRowKey 设计决定查询性能结果写 HBase 是整个链路里影响后续查询性能最关键的一步。RowKey 设计不好写入热点、查询慢、region 频繁分裂报表接口全卡。先看写入代码的骨架import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Put} import org.apache.hadoop.hbase.util.Bytes import org.apache.hadoop.hbase.{HBaseConfiguration, TableName} import org.apache.spark.sql.streaming.Trigger def buildRowKey(product: String, windowStart: Long, bucket: Int): String { val reverseTs (Long.MaxValue - windowStart).toString f$bucket%02d_${product}_$reverseTs } reportDF.writeStream .foreachBatch { (batchDF, batchId) batchDF.foreachPartition { rows val conn ConnectionFactory.createConnection(HBaseConfiguration.create()) val table conn.getTable(TableName.valueOf(insurance_report)) rows.foreach { row val win row.getAs[org.apache.spark.sql.Row](window) val windowStartTs win.getTimestamp(0).getTime val rowKey buildRowKey( row.getAs[String](product_type), windowStartTs, (windowStartTs % 97).toInt ) val put new Put(Bytes.toBytes(rowKey)) put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(premium_total), Bytes.toBytes(row.getAs[Double](premium_total))) put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(claim_count), Bytes.toBytes(row.getAs[Long](claim_count))) put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(claim_amount_total), Bytes.toBytes(row.getAs[Double](claim_amount_total))) table.put(put) } table.close() conn.close() } } .trigger(Trigger.ProcessingTime(60 seconds)) .option(checkpointLocation, /data/checkpoint/insurance_report) .start()RowKey 的构造是这段代码的重点。如果把时间戳直接放 RowKey 开头同一分钟写入的数据会顺序落在同一个 region 上形成写热点——这是 HBase 最常见的翻车姿势。这里做了两件事第一bucket 取时间戳对 97 取模把同一分钟的数据打散到 97 个分桶写入压力分散到不同 region第二时间戳用 Long.MaxValue 减去原值再转字符串让新的时间戳排前面按时间倒序扫描时 HBase 的 range scan 效率最高。生产环境里桶数按 region 数量调一般是 region 数的整数倍。表结构就一个列族 cf列分别存 premium_total、claim_count、claim_amount_total。列族别建多了HBase 一个列族对应一个 MemStore列族太多内存开销大报表场景一个列族足够。trigger设 ProcessingTime 60 秒控制结果写 HBase 的节奏跟下游报表刷新频率对齐。5. 避坑指南保险实时项目最容易翻车的六个点这套链路我前后跑过三遍才跑顺下面这些坑每个都真实发生过。按现象、原因、解决三步写清楚你照着排查就行。5.1 Executor 频繁 OOM任务无限重启现象Spark UI 里 executor 一个接一个挂掉日志报 java heap space任务重启后过一会儿又挂形成死循环。原因Kafka 里积压了大量事件maxRatePerPartition 没设或者设得太大executor 一次性拉了几十万条进内存窗口聚合的中间状态也全堆在堆内spark 内存分配根本扛不住。解决给每个分区限速spark.streaming.kafka.maxRatePerPartition从 500 开始调窗口聚合的状态用独立的 state store 目录别和 shuffle 目录挤在一起。血泪经验先限速再调内存顺序别搞反。一上来就加 executor memory治标不治本积压还在迟早再炸。5.2 报表数字和业务系统对不上现象报表显示今天保费收入 1200 万业务系统自己的统计是 1250 万差值看起来没规律今天多明天少。原因Kafka 生产端重试导致事件重复加上 Spark 任务重启后从 checkpoint 恢复时部分 batch 重复消费而结果写入 HBase 不是幂等的同一笔保费被累加了两次。解决三层防护。第一层生产端每条事件带全局唯一 event_id第二层HBase 写入前用 event_id 做去重或者把累加逻辑改成 HBase 的 increment 操作而不是 put 覆盖第三层checkpoint 用独立目录别让两个环境共用。这三层缺一层数字都对不齐。5.3 HBase 写热点一个 region 被打满现象HBase 监控里某个 region 的写请求数是其他 region 的几十倍region 频繁分裂写入延迟飙高报表查接口跟着变慢。原因RowKey 用了时间戳开头或者产品类型开头。时间戳顺序写必然打到一个 region产品类型只有三种热点更明显。这是 RowKey 设计问题不是 HBase 本身的问题。解决RowKey 加盐分桶时间戳反转把热门 key 打散第 4 章 buildRowKey 里已经给了实现。桶数别拍脑袋定按 region 数来上线后观察写入分布再调。加了盐之后 range scan 可能受影响所以加盐字段要放在 RowKey 低位时间戳反转放高位保证按时间查询还是高效的。5.4 窗口统计结果偏少数据“丢了”现象报表统计的理赔笔数明显少于业务系统但 Kafka 里数据一条没少Kafka 积压也是正常的。原因事件乱序。业务系统的事件到达 Kafka 的时间可能比事件产生时间晚几十秒甚至几分钟窗口按 event_time 聚合时晚到的数据落进了上一个窗口但那个窗口已经触发输出并关闭了没有 watermark 兜底数据就被丢了。解决withWatermark 设置合理的延迟阈值保险场景 60 秒起步同时打开 allowedLateness让晚到数据能触发窗口重新计算输出更新后的结果。这里要区分watermark 决定数据还等不等allowedLateness 决定晚到数据来了之后要不要重算两个配合着用才算完整。5.5 JSON schema 一变整批数据全变 null现象某天开始报表统计指标全是 0或者某一列全是空日志里 from_json 解析看起来没报错但数据结果不对。原因业务系统给事件里加了字段或者把某个字段类型改了比如 premium 从数字变成字符串。from_json 按旧 schema 解析新数据字段对不上返回 null聚合结果自然全错。这个坑最阴险因为它不报错只是静默变 null。解决schema 演进要跟着业务走解析层加一层校验——统计 json_str 里字段解析失败的条数超过阈值就报警同时在 from_json 外面包一层 try 语义脏数据单独写到一个 dead-letter topic别让脏数据混进主链路。别小看这个保险业务系统上线后改字段是常态不加这层校验等于埋雷。5.6 本地跑得好好的集群上任务起不来现象代码本地 IDEA 里跑通过打成 jar 丢到集群上启动报 ClassNotFoundException或者 JSON 解析行为跟本地完全不一样。原因八成是依赖冲突。本地用的是新版本 Kafka client集群上 Spark 自带的 kafka-clients 版本旧运行时类加载用的是集群那份另外 Scala 版本不一致也会出这种问题。解决打 fat jar 时把 spark-sql-kafka-0-10 的传递依赖排除掉尤其是 kafka-clients用集群自带的版本Spark 用 2.4 就配 Scala 2.11 的包用 3.x 就配 Scala 2.12 或 2.13别混。部署前先在集群上跑一个最小消费任务验证环境再上完整代码别把排查时间浪费在环境问题上。6. 上线前的验证与监控从「能跑」到「能上线」的最后一公里代码跑通只是第一步真正上线前我一般会花半天时间做三件事对账、压延迟、盯监控。这三件事做完才敢说这套实时链路是可靠的。对账是验证数据准确性的硬办法。挑一个时间窗口把 Kafka 里该窗口的事件一条条数出来跟 HBase 里对应窗口的聚合结果做对比。保费总额可以从事件明细累加验证理赔笔数直接数 claim 事件条数。两边对不上就回到第 5 章的幂等和乱序问题逐项排查。这个步骤我每次上线都做不做的话报表上线第二天业务方准来问数字对不对。压延迟是验证实时性的关键。给事件打上生产时间戳在 HBase 写入列里记录处理时间戳两者差值就是端到端延迟。保险报表场景目标一般是分钟级如果发现延迟超过 3 分钟先看 Kafka 积压量再看 Spark UI 的 batch 处理时间是不是逼近 batch 间隔最后检查 HBase 写入是否有热点。指标怎么看阈值参考batch 处理时间Spark UI Structured Streaming 页低于 batch 间隔的 60%Kafka 积压GetOffsetShell 的 latest 减 committed持续增长才报警端到端延迟处理时间戳减事件时间戳目标分钟级GC 频率jstat -gcutilFull GC 接近 0盯监控建议把 Spark UI 的 Structured Streaming 页面作为默认入口batch 处理时间、输入行数、watermark 水位这几个指标一眼就能看到瓶颈。内存和线程层面的问题用 spark 内存线程监测工具查——jstat 看老年代和 GC 频率jstack 看是不是有线程卡在 HBase 的 RPC 上。GC 频繁基本是状态太大或者限速不够线程卡 RPC 基本是 HBase 热点写或者连接池泄漏方向很明确。有一个习惯我一直保留着任何实时任务上线之前强制走一遍“从埋点到报表”的完整对账流程不管这个任务改了几个字段、动了哪个窗口参数。因为实时链路里每个环节都会不定时出幺蛾子Kafka 会重复投递Spark 会重启HBase 会分裂谁都不能保证稳定性只有对账能兜底。这套保险实时项目跑通之后我把同样的流程搬到了另外两个业务线把对账和监控脚本直接复用省掉了大量排查时间。希望帮到你。本文还有配套的精品资源点击获取
)
