简介：这份PDF技术文档聚焦Apache Flink与HBase在阿里巴巴电商业务中的落地实践，面向大数据开发工程师、实时计算架构师及对电商实时数据处理感兴趣的技术人员，帮助读者理解亿级数据量下流批一体与分布式存储的协同方案。资源包共1个PDF文件，大小约2.9MB，内容以技术讲解与代码示例为主，涵盖业务背景、典型场景、技术架构、具体实现与优化策略等模块。文档结合报表监控、商品库管理、用户足迹分析、生意参谋、供应链预警及全链路debug平台等真实场景，展开Flink流处理、HBase存储、Datahub接入与SQL/Table API的配合方式，并给出groupBy聚合、writeToHbaseSink写入、HBase表DDL定义及changelog表创建等实现细节。目前已有167人学习，适合希望掌握实时数据处理架构与排错思路的读者参考。

1. Flink+HBase 在电商实时链路里到底扛了什么活

大促零点一过，订单、加购、库存扣减、物流轨迹像洪水一样涌进来，MySQL 单机写入开始抖动，离线 T+1 报表根本追不上运营改价和风控拦截的节奏。Flink+HBase 这套组合在电商业务里的定位，说白了就是「实时计算 + 实时随机读写存储」：Flink 负责把埋点、Binlog、消息队列里的流按事件时间做窗口聚合和状态计算，HBase 负责把计算结果按行键落到一张能扛百万级 QPS 随机读写的宽表里，供风控、推荐、实时大屏去查。它适合的是已经有一定数据规模、离线链路开始拖后腿的团队，不是玩具项目。下面按「选型理由 → 环境搭建 → 写入实现 → 避坑 → 调优验证」推一遍，能照着复现。

2. 为什么电商实时场景选 Flink 配 HBase 而不是别的

2.1 从业务诉求反推存储选型

电商实时链路里最典型的三个诉求：一是按用户维度查最近行为，比如风控要判断这个账号 5 分钟内是否换了 3 个收货地址；二是按商品维度做实时累计，比如某 SKU 的分钟级销量要立刻反映到库存预警；三是写入要能扛住大促峰值，读延迟要稳定在个位数毫秒。这三条决定了存储必须支持按主键随机读写、水平扩展、写入不依赖复杂事务。

常见做法是拿 HBase 和几个候选对比。MySQL 分库分表能扛写入，但按用户+时间做范围扫描时二级索引维护成本高，扩容要停机迁移；Redis 读快，但全量行为数据放内存成本顶不住，持久化也不是为海量宽表设计的；ClickHouse 适合 OLAP 聚合查询，但按单行主键高频点查不是它的强项，更新删除也偏重。HBase 的 RowKey 设计天然支持「用户ID+时间戳」这种前缀扫描，写入走 LSM 树顺序落盘，Region 自动分裂，正好对上电商的读写模式。

Flink 这边选它的理由更直接：电商流数据乱序严重，埋点上报延迟从几百毫秒到几分钟都有，Flink 的 Event Time + Watermark 机制能把乱序数据按事件真实发生时间归到正确窗口，这是 Spark Streaming 微批模型做起来更别扭的地方。加上 Flink 的状态后端可以存几 TB 的 KeyedState，做去重、会话窗口、CEP 风控规则都够用。

2.2 版本与依赖怎么定

版本这块是血泪经验，Flink 和 HBase 的版本兼容表一定要先查。Flink 1.14 之后flink-connector-hbase拆成了独立模块，和 HBase 2.x 的对应关系比较清晰；如果还在用 Flink 1.13 以前，连接器是内置在flink-connector-hbase里的，API 不一样。HBase 侧建议 2.4.x 或 2.5.x，1.x 在老集群里还有，但新项目没必要。

依赖引入时注意flink-connector-hbase-2.2这个 artifact 名字里的 2.2 指的是 HBase 2.2+ 兼容，不是只能配 2.2。Maven 里还要把 HBase 的hbase-client、hbase-common一起带上，否则运行时报NoClassDefFoundError。

<!-- pom.xml 关键依赖，scope 用 provided 还是 compile 看集群是否自带 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-hbase-2.2</artifactId> <version>1.17.2</version> </dependency> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-client</artifactId> <version>2.4.17</version> </dependency>

参数说明：flink-connector-hbase-2.2的版本号跟 Flink 主版本走，不是跟 HBase 走；hbase-client版本要和集群服务端一致，差一个小版本可能出RpcRetryingCaller超时。如果集群已经带了 HBase 依赖，把 scope 改成provided避免类冲突。

3. 从零把 Flink 写 HBase 的最小链路跑通

3.1 HBase 表设计与 RowKey 怎么定

电商场景最常踩的坑就是 RowKey 设计。假设要做「用户实时行为宽表」，一张表存用户最近的操作。RowKey 如果直接用userId，热点问题严重，大 V 用户会把单个 Region 打爆；如果直接用时间戳+userId，按用户查就要全表扫。

常见做法是salting + 反转 + 业务前缀组合。比如 RowKey 设计成hash(userId)%16 + userId反转 + 时间戳倒序。加盐把写分散到 16 个 Region，反转 userId 让相似 ID 不连续，时间戳倒序让最新数据排在最前，查最近 N 条直接scan前几行就够。

# 建表，预分区 16 个，列族按访问模式拆 create 'user_behavior', {NAME => 'info', VERSIONS => 1, BLOOMFILTER => 'ROW'}, \ {NAME => 'stat', VERSIONS => 1, COMPRESSION => 'SNAPPY'}, \ SPLITS => ['0','1','2','3','4','5','6','7','8','9','a','b','c','d','e','f']

参数说明：VERSIONS => 1是因为电商行为表通常只保留最新值，多版本会撑大存储；BLOOMFILTER => 'ROW'对随机点查有加速；COMPRESSION => 'SNAPPY'在 CPU 和压缩比之间平衡，写入吞吐影响小。预分区数量按 RegionServer 数量乘以 2 到 4 来估，16 个是单机测试值，生产要按集群规模调。

3.2 Flink 侧读取 Kafka 并做窗口聚合

上游一般是 Kafka 里的埋点或 Binlog。Flink 消费后先做keyBy(userId)，再用TumblingEventTimeWindows做分钟级聚合，把结果写到 HBase。这里要注意 Watermark 的延迟设置，电商埋点乱序常见 30 秒到 2 分钟，forBoundedOutOfOrderness设太小会丢数据，设太大窗口触发慢。

// Flink 主流程：Kafka -> 窗口聚合 -> HBase Sink StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); // 1.12 之前写法，新版用 WatermarkStrategy DataStream<BehaviorEvent> source = env.addSource( new FlinkKafkaConsumer<>("behavior_topic", new BehaviorSchema(), kafkaProps)) .assignTimestampsAndWatermarks( WatermarkStrategy.<BehaviorEvent>forBoundedOutOfOrderness(Duration.ofSeconds(60)) .withTimestampAssigner((e, ts) -> e.getEventTime())); DataStream<UserStat> stat = source .keyBy(BehaviorEvent::getUserId) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new BehaviorAggregator()); stat.addSink(new HBaseSink()); // 自定义 Sink，见下节

逻辑说明：forBoundedOutOfOrderness(60秒)表示允许数据迟到 60 秒，超过的走 side output 单独处理；keyBy(userId)保证同一用户的数据进同一个算子实例，状态本地化；aggregate比apply省内存，增量聚合不缓存窗口内全量元素。参数上，窗口大小按业务定，风控常用 1 分钟，大屏常用 5 秒滑动窗口。

3.3 自定义 HBase Sink 的写入实现

Flink 自带的HBaseSink在flink-connector-hbase里有，但生产上更常见的是自己写RichSinkFunction，因为要控制批量提交、缓冲大小和失败重试。下面是一个最小可用版本。

public class HBaseSink extends RichSinkFunction<UserStat> { private transient Connection conn; private transient BufferedMutator mutator; private static final int BUFFER_SIZE = 2 * 1024 * 1024; // 2MB 缓冲 @Override public void open(Configuration params) throws Exception { Configuration conf = HBaseConfiguration.create(); conf.set("hbase.zookeeper.quorum", "zk1,zk2,zk3"); conf.set("hbase.zookeeper.property.clientPort", "2181"); conn = ConnectionFactory.createConnection(conf); BufferedMutatorParams mp = new BufferedMutatorParams(TableName.valueOf("user_behavior")) .writeBufferSize(BUFFER_SIZE); mutator = conn.getBufferedMutator(mp); } @Override public void invoke(UserStat stat, Context ctx) throws Exception { String rowKey = buildRowKey(stat.getUserId(), stat.getWindowEnd()); Put put = new Put(Bytes.toBytes(rowKey)); put.addColumn(Bytes.toBytes("stat"), Bytes.toBytes("cnt"), Bytes.toBytes(stat.getCount())); put.addColumn(Bytes.toBytes("stat"), Bytes.toBytes("last_time"), Bytes.toBytes(stat.getWindowEnd())); mutator.mutate(put); // 异步缓冲，满 2MB 自动 flush } @Override public void close() throws Exception { if (mutator != null) mutator.close(); // close 会 flush 剩余缓冲 if (conn != null) conn.close(); } private String buildRowKey(String userId, long ts) { int salt = Math.abs(userId.hashCode()) % 16; String reversed = new StringBuilder(userId).reverse().toString(); return String.format("%x_%s_%d", salt, reversed, Long.MAX_VALUE - ts); } }

逻辑说明：BufferedMutator是 HBase 客户端推荐的批量写入方式，比每次Table.put少很多 RPC；writeBufferSize设 2MB 是经验值，太小 RPC 多，太大失败重放代价高。buildRowKey里Long.MAX_VALUE - ts实现时间倒序，查最新数据时scan从表头开始即可。参数上，hbase.zookeeper.quorum要填全，少一个节点在 ZK 选举时会连不上。

注意：close()里必须先关mutator再关conn，顺序反了缓冲里的数据会丢。这是我在压测时丢过一批数据才记住的。

4. 写入 HBase 时最容易翻车的几个点

4.1 现象：RegionServer 频繁 GC，写入延迟飙升

原因通常是单次Put太大或者批量提交条数太多，导致 MemStore 快速膨胀触发 flush，RegionServer 堆内存吃紧。电商行为表一条记录如果塞了几十个字段，单行能到几 KB，批量 1000 条就是几 MB。

解决：单行控制在 1KB 以内，大字段拆到独立列族或独立表；批量提交条数按writeBufferSize反推，2MB 缓冲配 500 到 1000 条比较稳；RegionServer 堆内存建议不低于 16GB，hbase.regionserver.global.memstore.size保持默认 0.4，不要为了写入调太大，否则读缓存被挤掉。

4.2 现象：RowKey 热点，单个 Region 请求量是其他 Region 的几十倍

原因就是 RowKey 单调递增或者集中在少数前缀。比如直接用时间戳开头，所有写入都打到最后一个 Region。

解决：加盐前缀，hash(userId)%N是最简单的；如果业务允许，用MD5(userId)前几位做前缀也行。加盐后查询要按盐值遍历，比如查某用户数据要扫 16 个 Region，这是代价，所以盐值数量别设太大，16 到 32 够用。另外预分区要配合盐值范围，建表时 SPLITS 要覆盖所有盐值前缀。

4.3 现象：Flink 任务反压，Checkpoint 超时失败

原因一般是 Sink 写入慢拖累了整个链路，或者 Checkpoint 时BufferedMutator缓冲没 flush，状态快照和实际写入不一致。

解决：给 Sink 加独立的线程池或者用AsyncSink模式，别让写入阻塞算子线程；Checkpoint 前手动mutator.flush()，在snapshotState里做；调大execution.checkpointing.timeout，但根本还是解决写入瓶颈。另外execution.checkpointing.unaligned在反压严重时能救急，但会增大状态。

4.4 现象：HBase 客户端报RpcRetryingCaller: Call exception, tries=10

原因通常是 ZK 连接不稳、RegionServer 负载过高或者hbase.rpc.timeout设太短。电商大促时 RegionServer 请求队列满，RPC 排队超时很常见。

解决：hbase.rpc.timeout从默认 60 秒适当调大，但别超过业务容忍度；hbase.client.retries.number默认 15 次，可以保持；关键是监控 RegionServer 的callQueueLength，持续大于 100 就要加节点或优化 RowKey。ZK 侧检查zookeeper.session.timeout和集群网络延迟。

4.5 现象：Flink 作业重启后 HBase 里出现重复数据

原因是 Sink 不是幂等的，Checkpoint 恢复后从上次位点重放，同一批数据写了两次。HBase 的Put默认是覆盖，但如果 RowKey 里带了随机数或者时间戳精度不够，就会产生两行。

解决：RowKey 必须由业务主键+窗口时间唯一确定，不能带随机因子；如果业务允许，用checkAndPut做幂等，但性能会降；更常见的做法是下游查询时按版本或时间去重，或者接受最终一致。这个问题没有银弹，设计阶段就要想清楚。

5. 调优参数与验证方法：怎么确认这套链路真的扛得住

5.1 写入侧必调的四个参数

HBase 写入性能对参数敏感，下面四个是我每次上线前必看的。

调write.buffer时注意，缓冲越大，失败重放的数据量越大，要配合重试策略。flush.size调大能减少 flush 次数，但 RegionServer 重启恢复时间变长。max.filesize调大适合写入量大、Region 数多的集群，但单个 Region 太大影响负载均衡。

5.2 用压测验证而不是拍脑袋

上线前至少做两轮压测：一轮纯写入，一轮混合读写。纯写入用YCSB或者自己写个多线程客户端，目标 QPS 按大促峰值的 1.5 倍估。混合读写按 7:3 或 8:2 的比例，读用Get和Scan各占一半。

# YCSB 压测 HBase 写入，先建好 usertable bin/ycsb load hbase2 -P workloads/workloada -p table=user_behavior \ -p columnfamily=stat -p recordcount=10000000 -threads 64

参数说明：recordcount按实际数据量估，threads按客户端机器核数乘 2 到 4。压测时盯三个指标：RegionServer 的writeRequestCount、memStoreSize、callQueueLength。callQueueLength持续超过 100 说明写入已经排队，要加 RegionServer 或优化 RowKey。

5.3 一个容易被忽略的验证点：数据一致性

Flink 写 HBase 后，怎么确认数据没丢没重？我的习惯是跑一个对账任务：从 Kafka 原始 topic 按窗口重新聚合一份结果，和 HBase 里的数据按 RowKey 比对。差异行数超过万分之一就要查。对账任务不用实时，T+1 跑一次就行，但能兜住大部分写入逻辑 bug。

另外 Checkpoint 恢复测试必须做：手动 kill TaskManager，看作业恢复后 HBase 里的数据是否和 Kafka 位点对齐。这个测试能暴露 Sink 幂等性和 Checkpoint 配置的问题，比看日志管用。

5.4 最后说一个我踩过的坑

早期做的时候，我把 HBase Sink 的open方法里创建连接写成了每次invoke都创建，压测时 QPS 上不去，查了半天才发现是连接泄漏。后来改成open里创建、close里释放，QPS 直接翻了 10 倍。这个习惯我一直保持到现在：任何外部连接，生命周期必须和算子实例对齐，不能和单条数据对齐。希望帮到你。

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