Flink HBase SQL Connector RowKey 设计、Upsert 语义、维表 Join、缓存与写入调优

1. 先把语义说死：HBase Connector 永远是 Upsert

HBase 在 Flink SQL 里有一个非常“硬”的特性：

• HBase 总是 Upsert 模式（交换 changelog：UPDATE/DELETE/INSERT 都能处理）
• 表里必须有一个rowkey 字段
• PRIMARY KEY 必须定义在 rowkey 上；如果你不写 PRIMARY KEY，默认把 rowkey 当主键

你可以把它理解成：Flink 写 HBase，本质就是在维护一个“按 rowkey 定位的宽表”。

2. 映射规则：列族必须用 ROW，rowkey 是原子字段

HBase 的模型是rowkey + family + qualifier + value，Flink SQL 的映射规则对应得很直：

• 每个列族必须声明为ROW<...>

  • ROW 字段名 = family 名
  • ROW 内嵌字段名 = qualifier 名

• 除 ROW 类型之外的单一原子类型字段（如 STRING/BIGINT/INT）会被识别为rowkey

• 不需要把所有列族/qualifier 都声明出来，用到什么声明什么

建表模板（你给的官方示例风格）：

CREATETABLEhTable(rowkeyINT,family1ROW<q1INT>,family2ROW<q2 STRING,q3BIGINT>,family3ROW<q4DOUBLE,q5BOOLEAN,q6 STRING>,PRIMARYKEY(rowkey)NOTENFORCED)WITH('connector'='hbase-2.2','table-name'='mytable','zookeeper.quorum'='localhost:2181');

提醒两点：

• rowkey 字段名随便起，但如果是关键字要用反引号
• table-name默认 namespace 是default，指定 namespace 用ns:table

3. 写入方式：用 ROW(…) 构造列族

写 HBase 时，每个列族要传一个 ROW 值：

INSERTINTOhTableSELECTrowkey,ROW(f1q1),ROW(f2q2,f2q3),ROW(f3q4,f3q5,f3q6)FROMT;

把它记成一句话：HBase 的 family 在 Flink 里是“一个结构体”。

4. 读取与维表 Join：Scan + Temporal Join（Lookup）

4.1 普通读取（Scan）

SELECTrowkey,family1,family3.q4,family3.q6FROMhTable;

4.2 作为维表做 Temporal Join（经典用法）

SELECT*FROMmyTopicLEFTJOINhTableFORSYSTEM_TIMEASOFmyTopic.proctimeONmyTopic.key=hTable.rowkey;

维表查询默认是同步的，HBase 作为外部系统会直接影响算子吞吐。你可以启用异步 Lookup（仅hbase-2.2支持）：

'lookup.async'='true'

5. 生产级的关键：RowKey 设计与热点规避

HBase 写入性能的天花板，很多时候不是 Flink，也不是网络，而是RowKey 分布。

5.1 两个“必踩坑”的 RowKey

1）递增 ID / 时间戳在尾部

• rowkey = userId（userId 可能递增/集中）
• rowkey = 20260122123456（时间单调递增）
  结果：写入集中打到最后几个 Region，热点爆炸，吞吐塌方。

2）时间序列直接正序

• rowkey = deviceId_yyyyMMddHHmmss
  如果 deviceId 少且上报密集，照样热点。

5.2 通用解法：Hash/Salt 前缀，让写入均匀

最常用的做法是给 rowkey 加一个固定桶数的前缀：

• 前缀 =SUBSTRING(MD5(key), 1, 2)（256 桶）
• 或者前缀 =MOD(hash, 16)（16 桶）

Flink SQL 里你可以这样构造（示意）：

SELECTCONCAT(SUBSTRING(MD5(user_id),1,2),'_',user_id)ASrowkey,...FROMsrc;

这样同一个 user_id 永远落在同一个桶前缀下，既能均匀写入，又能稳定定位。

桶数怎么选：

• Region 少、并发低：16/32 桶
• 并发高、Region 多：64/128/256 桶
  桶数太大也会让 Scan 更碎，按实际读写比例取舍。

5.3 时间序列更狠一点：倒排时间 + 分桶

想做“最近数据优先读”，又想避免热点，可以用：

• salt + reverse_ts + deviceId
  reverse_ts = 把时间戳做倒排，越新越小，范围 scan 更友好

示意：

SELECTCONCAT(SUBSTRING(MD5(device_id),1,2),'_',LPAD(CAST(9999999999999-ts_msASSTRING),13,'0'),'_',device_id)ASrowkey,...FROMsrc;

这套组合拳适合：

• 海量设备日志
• 以时间范围查询为主
• 需要按设备维度定位

6. 一致性与幂等：Checkpoint 重放、乱序覆盖、NULL 抹数据

你要把这个事实刻进 DNA：
Flink 开了 Checkpoint 后，失败恢复会从上一次成功 checkpoint 重新处理，因此对外部系统通常是至少一次。

HBase 为什么还经常“看起来像 exactly-once”？因为它是 Upsert：

• 同一个 rowkey、同一个 qualifier，重复写通常会覆盖成同一个最终值
• 这在很多“最终态宽表”场景非常好用

但注意三个坑：

6.1 乱序更新：晚到数据可能覆盖新值

如果你做的是用户画像/状态表，事件可能乱序：

• 新事件先到，写入 status=NEW
• 旧事件后到，又写入 status=OLD
  最终 HBase 里变旧了

解决思路（选一种）：

• 上游先做按主键去重/保序（例如按事件时间取最新）
• 把事件时间也写到 HBase（例如写last_event_time），下游读取时以时间判断
• 把时间作为 HBase 的版本（需要更深度的版本控制策略）

6.2 NULL 覆盖：最隐蔽的“抹数据”事故

HBase connector 的编码规则里：

• 除 STRING 外，null 会编码为空 bytes
  这意味着：你上游某字段变成 null，写入可能把 HBase 原有字段“覆盖成空/NULL”。

生产里通常更想要的是：字段为 null 就别写，别覆盖。
这时候用：

'sink.ignore-null-value'='true'

或者在 SQL 里显式控制：

• 对不该被抹的字段，用COALESCE(newVal, oldVal)（需要能拿到 oldVal）
• 或者直接过滤掉 null 更新

6.3 STRING 的 null 规则：null-string-literal 很关键

STRING 是例外：空 bytes 解码成什么，由null-string-literal决定（默认是"null"）。
如果你的业务字段里可能真的出现字符串"null"，强烈建议改成一个不可能出现的字面量：

'null-string-literal'='__HBASE_NULL__'

7. 写入调优三件套：max-rows / max-size / interval

写 HBase 追求吞吐，核心就是批量 flush：

• sink.buffer-flush.max-rows（默认 1000）
• sink.buffer-flush.max-size（默认 2mb）
• sink.buffer-flush.interval（默认 1s）

原则：

• 吞吐优先：适当增大 rows/size，interval 可以略大
• 延迟优先：interval 调小，rows/size 控制住

示例（偏吞吐）：

WITH('sink.buffer-flush.max-rows'='2000','sink.buffer-flush.max-size'='4mb','sink.buffer-flush.interval'='1s')

别忘了一个现实：flush 变慢会直接拖长 checkpoint（反压也更明显），所以需要结合集群状态逐步调，不要一步到位拉满。

8. Lookup 性能：缓存 PARTIAL + 异步 lookup（两把刀）

8.1 PARTIAL 缓存（TaskManager 进程级别）

'lookup.cache'='PARTIAL','lookup.partial-cache.max-rows'='200000','lookup.partial-cache.expire-after-write'='10 min','lookup.partial-cache.expire-after-access'='5 min','lookup.partial-cache.caching-missing-key'='true'

经验：

• 热点维表非常吃缓存，提升巨大
• TTL 越短越准，越长越省 IO
• caching-missing-key=true能抗“热点 miss”，但如果 key 会新增，可能短时间查不到（被缓存住）

8.2 异步 lookup（hbase-2.2 专属）

'lookup.async'='true'

适合：

• 维表请求 RT 不稳定
• 并发 lookup 多
• 不希望维表拖垮主链路

不适合：

• 你对单条记录处理严格保序（需要额外设计）
• 外部系统扛不住更高并发（异步会把并发放大）

9. HBase 参数透传：properties.*（Kerberos/超时/重试都靠它）

HBase 生产绕不开各种 client 配置，connector 提供了透传方式：

'properties.hbase.security.authentication'='kerberos'

你可以把常见的超时、重试、RPC 配置都透传进去（按你们集群策略来），Flink 会去掉properties.前缀传给底层 HBase Client。

10. 一套可直接复用的生产 DDL 模板

CREATETABLEuser_profile_hbase(user_id STRING,baseROW<name STRING,ageINT,statusBOOLEAN>,statROW<uvBIGINT,pvBIGINT>,PRIMARYKEY(user_id)NOTENFORCED)WITH('connector'='hbase-2.2','table-name'='ns:user_profile','zookeeper.quorum'='zk1:2181,zk2:2181,zk3:2181','zookeeper.znode.parent'='/hbase','null-string-literal'='__HBASE_NULL__','sink.ignore-null-value'='true','sink.buffer-flush.max-rows'='2000','sink.buffer-flush.max-size'='4mb','sink.buffer-flush.interval'='1s','lookup.async'='true','lookup.cache'='PARTIAL','lookup.partial-cache.max-rows'='200000','lookup.partial-cache.expire-after-write'='10 min','lookup.partial-cache.caching-missing-key'='true');

11. 避坑清单（上线前对一下，能省很多夜宵）

• RowKey 是否会单调递增导致热点？是否加了 salt/hash 前缀？
• 是否存在乱序更新？晚到数据会不会覆盖新值？
• 是否允许 NULL 覆盖旧值？若不允许，是否设置了sink.ignore-null-value=true？
• STRING 的 null literal 是否会跟业务值冲突？是否设置了null-string-literal？
• 维表 Join 是否被 HBase RT 拖垮？是否启用 async lookup/partial cache？
• flush 参数是否与 checkpoint 周期匹配？是否会把 checkpoint 拉爆？