企业官网建设流程全解析

1. 这不是“又一个ETL流程”，而是实时数仓链路里最易被低估的毛细血管

我第一次在某电商中台项目里看到用maxwell+kafka+thrift这套组合把 MySQL 订单库同步进 HBase 时，以为只是个临时过渡方案。结果上线三个月后，下游所有实时风控、用户行为画像、库存预占服务都悄悄切到了这条链路上——不是因为快，而是因为它稳得让人忘记它的存在。

你可能已经听过太多“MySQL → Kafka → Flink → HBase”这种标准答案，但现实里，当你的团队没有专职实时计算工程师、Flink 集群资源吃紧、又必须在两周内交付一个支持毫秒级查询的用户标签宽表时，maxwell+kafka+thrift就成了那个不炫技却真正扛住日均 800 万变更事件的“沉默管道”。

它不解决“复杂计算”，也不做“多源关联”，它只干一件事：把 MySQL 的每一行 INSERT/UPDATE/DELETE，以确定性、低延迟、可重放的方式，原样喂给 HBase。而这个“原样”，恰恰是很多团队踩坑的起点——他们误以为只要数据进了 Kafka 就算成功，却没意识到：从 binlog event 到 HBase Put 请求之间，藏着至少 5 层语义转换和 3 类典型失真风险。

关键词里虽然没写，但实际落地中，你绕不开的其实是这四个硬核问题：

• maxwell 如何精准捕获 DDL 变更并动态适配 HBase 表结构？（不是所有字段都能直接映射）
• kafka 消息体该用 JSON 还是 Thrift 序列化？为什么我们最终弃用 JSON 而坚持用 .thrift 文件定义 schema？
• thrift 接口层如何设计才能让 HBase 写入既支持高吞吐又避免 RegionServer OOM？（别小看一个 putBatch 的 batch size）
• 当 MySQL 主从切换或 maxwell 实例宕机时，如何保证 HBase 数据与源库最终一致，且不依赖人工干预？

这不是一个“装完就能跑”的玩具链路。它是一条需要你亲手校准每个齿轮咬合间隙的工业级数据通道。接下来，我会按真实项目推进顺序，带你一节一节拧紧这些螺丝——不讲原理图，只说哪颗螺丝拧太紧会崩，哪颗拧太松会漏。

2. Maxwell 不是 binlog 解析器，它是 MySQL 与下游系统的“语义翻译官”

很多人把 maxwell 当成一个“MySQL binlog 到 JSON 的转换器”，这是对它能力的最大误判。它真正的价值，在于把 MySQL 的物理变更事件（row-based binlog），翻译成下游系统能理解的、带业务语义的领域事件。而这个翻译过程，恰恰是整条链路稳定性的第一道闸门。

2.1 为什么不能直接消费 raw binlog？三个血泪教训

某次大促前压测，我们曾尝试跳过 maxwell，用 canal 直接解析 binlog 并发往 Kafka。结果在凌晨三点收到告警：HBase 中出现大量user_id=0的脏数据。排查发现，MySQL 中有个tinyint(1)字段实际存的是布尔值（0/1），但 canal 默认将其转为字符串"0"或"1"，而我们的 HBase 写入逻辑误判为整型0，导致所有is_vip=1的用户被写成is_vip=0。

maxwell 的设计哲学完全不同：它在启动时主动连接 MySQL，读取 information_schema 获取每张表的真实列类型、是否允许 NULL、默认值、字符集。这意味着：

• TINYINT(1)被识别为TINYINT，而非字符串；
• DATETIME字段自动带上时区信息（如2024-06-15 14:23:00+08:00），避免跨时区写入偏差；
• ENUM('active','inactive')字段在 JSON 输出中明确标注"type": "enum", "values": ["active","inactive"]，下游可据此做合法性校验。

提示：maxwell 的--output_ddl参数必须开启。它会在 Kafka 中发送特殊 topic（默认maxwell.ddl）的消息，内容是完整的CREATE TABLE或ALTER TABLE语句。这不是可选项——它是后续 thrift schema 动态演化的唯一依据。

2.2 配置文件里的 4 个关键参数，决定你能否活过第一个主从切换

maxwell 的config.properties看似简单，但以下 4 个参数若配置不当，会在主从切换时引发雪崩：

特别强调output_binlog_position：它输出的不是简单的数字，而是形如{"file":"mysql-bin.000234","position":123456789,"gtid":"3e11fa47-71ca-11e1-9e33-c80aa9429562:1-5"}的结构体。这个结构，是我们后续在 thrift 层做幂等写入和断点续传的唯一锚点。

2.3 实战经验：如何让 maxwell 自动适应 DDL 变更而不中断？

某次运营同学紧急上线一个新字段user_profile.extra_info json，DBA 执行了ALTER TABLE user_profile ADD COLUMN extra_info JSON AFTER status。maxwell 默认行为是：检测到新字段后，将整条消息的data字段转为{"id":123,"name":"张三",...,"extra_info":null}—— 注意，是null，不是空对象{}。

这导致下游 thrift 解析时报NullPointerException。我们通过两个动作解决：

1. 启用--output_nulls=true（默认 false）：让 maxwell 显式输出null值，而非省略字段。这样下游 thrift 可以明确区分“字段不存在”和“字段值为 null”。

2. 在 maxwell 启动命令中加入--filter='exclude: *.*, include: user_db.user_profile'：精确控制监听范围。避免因监听全库导致 DDL 变更风暴（如某测试库频繁建表）拖垮 maxwell。

注意：maxwell 的 filter 是正则匹配，*.*表示所有库所有表。生产环境务必显式include关键表，exclude其他。我们曾因未加 filter，导致 maxwell 在凌晨自动扫描到一个 2TB 的历史归档库，内存暴涨至 8GB 后 OOM。

2.4 你必须自己写的监控脚本：3 行 bash 判断 maxwell 是否健康

光靠 maxwell 自带的/metrics端点不够。我们每天凌晨用以下脚本检查：

#!/bin/bash # check_maxwell_health.sh KAFKA_TOPIC="maxwell" LATEST_OFFSET=$(kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server kafka01:9092 --topic "$KAFKA_TOPIC" --time -1 | awk -F ':' '{print $3}') CURRENT_OFFSET=$(curl -s http://maxwell01:8080/metrics | grep 'kafka_producer_record_send_rate' | awk '{print $2}') if [ $(echo "$LATEST_OFFSET - $CURRENT_OFFSET" | bc) -gt 10000 ]; then echo "ALERT: maxwell lag > 10000 messages" exit 1 fi

这个脚本的核心逻辑是：对比 Kafka 中该 topic 的最新 offset 和 maxwell 当前已提交的 offset。差值超过 10000，说明 maxwell 写入 Kafka 严重滞后——这往往意味着 MySQL 网络抖动、Kafka 分区 Leader 切换，或是 maxwell 内存不足触发 Full GC。

3. Kafka 不是消息队列，它是实时数据链路的“压力缓冲池”与“语义校验墙”

把 Kafka 当作“消息中转站”是新手最大的认知陷阱。在这条链路里，Kafka 承担着三重不可替代的角色：流量削峰的缓冲池、序列化格式的强制校验墙、以及跨系统事务边界的天然分界线。而其中最容易被忽视的，是第二点——序列化格式的选择，直接决定了下游 thrift 层的健壮性。

3.1 JSON vs Thrift：为什么我们砍掉了所有 JSON Schema 校验代码？

项目初期，我们用 maxwell 默认的 JSON 输出，Kafka 消息体长这样：

{ "database": "user_db", "table": "user_profile", "type": "update", "ts": 1718467200, "xid": 123456789, "commit": true, "data": { "id": 1001, "name": "李四", "email": "lisi@example.com", "updated_at": "2024-06-15 14:23:00" }, "old": { "name": "李思", "email": "lisi_old@example.com" } }

看起来很清晰？问题出在data和old这两个嵌套对象上。当 MySQL 表新增字段phone VARCHAR(20)后，maxwell 会自动在data中加入"phone": null，但不会修改old对象结构。这就导致：

• 下游 Java 消费者用 Jackson 反序列化时，old对象缺少phone字段，反序列化失败；
• 若强行忽略缺失字段，old.phone值为null，但业务逻辑误以为这是“用户清空了手机号”，触发错误的风控规则。

我们试过用 JSON Schema 做强校验，但很快放弃——因为每次 DDL 变更都要手动更新 schema 文件，且无法处理old对象的动态字段缺失问题。

Thrift 的解法是降维打击：我们定义了一个固定的.thrift文件：

struct MaxwellEvent { 1: required string database; 2: required string table; 3: required string type; // "insert", "update", "delete" 4: required i64 ts; 5: optional i64 xid; 6: optional bool commit; 7: required map<string, string> data; // 所有字段转为 string 8: optional map<string, string> old; // 同上 9: required string binlog_file; 10: required i64 binlog_position; }

关键点在于第 7、8 行：map<string, string>强制所有字段值转为字符串。MySQL 的INT、DATETIME、JSON全部先 toString()，再塞进 map。这看似“丢精度”，实则是用空间换鲁棒性：

• old对象缺失字段？map.get("phone") 返回 null，业务逻辑可安全判断；
• data中updated_at是"2024-06-15 14:23:00"还是"1718467200000"？由下游 HBase 写入层统一解析，maxwell 层不参与；
• 新增字段？map 自动扩容，无需改 thrift 定义。

经验：Thrift 的map<string, string>是我们踩过 3 次坑后定下的铁律。它让 maxwell 和下游彻底解耦——maxwell 只负责“搬运字符串”，解析和类型转换交给更可控的 HBase 写入层。

3.2 Kafka Topic 设计：为什么我们坚持“一表一 topic”，哪怕要建 200+ 个？

常见做法是：所有 MySQL 表变更发到一个大 topic（如mysql-changes），下游用key字段（如database:table）做分区。但我们选择为每张业务表单独建 topic（user_db.user_profile、order_db.order_detail）。

理由很实在：

1. 流控隔离：订单库每秒 5000+ 更新，用户库只有 200 QPS。如果混在一个 topic，订单流量高峰会挤占用户数据的消费带宽，导致用户标签更新延迟。
2. 权限收敛：HBase 写入服务只需订阅自己关心的 topic，避免消费无关数据增加 GC 压力。
3. Schema 演化独立：user_profile加字段不影响order_detail的消费逻辑，无需版本兼容。

代价是运维成本上升。我们用 Ansible 自动化管理：

• 每张表的 topic 创建脚本存于kafka-topics.yml，包含--partitions 12（按 HBase region 数量对齐）、--replication-factor 3；
• maxwell 的filter配置与 topic 名称严格对应；
• 监控脚本按 topic 分组告警。

提示：topic 名称必须与 MySQL 表名完全一致（含下划线），这是后续 thrift 层自动路由到 HBase 表的关键。我们禁止任何别名或缩写。

3.3 消费者组的生死线：enable.auto.commit=false是唯一正确姿势

Kafka 消费者必须关闭自动提交 offset：

props.put("enable.auto.commit", "false"); props.put("auto.offset.reset", "earliest"); // 从头消费，用于灾备

原因在于：HBase 写入是外部系统调用，其成功与否无法由 Kafka broker 确认。如果 auto commit 开启，消费者拉到消息后立即提交 offset，但 HBase 写入因网络超时失败，这条消息就永久丢失。

我们的消费逻辑是原子的：

while (true) { ConsumerRecords<String, ByteBuffer> records = consumer.poll(Duration.ofMillis(100)); List<Put> puts = new ArrayList<>(); for (ConsumerRecord<String, ByteBuffer> record : records) { MaxwellEvent event = deserializeThrift(record.value()); // 反序列化 puts.add(buildHBasePut(event)); // 构建 HBase Put } try { hbaseTable.put(puts); // 批量写入 HBase consumer.commitSync(); // 仅当 HBase 写入成功，才提交 offset } catch (Exception e) { log.error("HBase write failed, will retry", e); // 不提交 offset，下次 poll 会重拉这批消息 } }

这个commitSync()调用，就是整条链路“至少一次”（at-least-once）语义的基石。它让 Kafka 成为真正的“重试保险丝”——只要 HBase 写入失败，消息就卡在 partition 里，直到写入成功。

3.4 你必须知道的 Kafka 参数陷阱：linger.ms和batch.size的黄金配比

maxwell 向 Kafka 生产消息时，linger.ms（等待更多消息打包的时间）和batch.size（批次大小）的设置，直接影响 HBase 写入吞吐。

我们经过 7 轮压测得出的结论：

关键经验：batch.size单位是字节，不是消息条数。一条典型的user_profileupdate 消息约 1.2KB，所以65536≈ 54 条消息。这个数字必须根据你表的平均消息大小动态调整，不能照搬。

4. Thrift 不是 RPC 框架，它是 HBase 写入层的“类型安全防火墙”与“业务逻辑收口”

很多团队把 thrift 当作“Java 调 HBase 的一层封装”，这是对它价值的严重矮化。在这条链路里，thrift 是唯一有权决定“什么数据能进 HBase”、“以什么格式进”、“进哪个 CF/CQ”的守门人。它把原本散落在各处的校验逻辑（空值检查、长度限制、枚举值白名单）全部收束到一个 .thrift 文件和一组生成的接口里。

4.1 Thrift IDL 设计：为什么我们拒绝string rowkey，而坚持binary rowkey？

初版 thrift 定义中，我们写了：

struct HBasePutRequest { 1: required string rowkey; // 错误！ 2: required string table_name; 3: required list<ColumnValue> columns; }

上线后发现严重问题：MySQL 的BIGINT主键（如1234567890123456789）转成 string 后，作为 HBase rowkey 会导致region 分布严重不均——所有大数值 rowkey 都落到同一个 region，热点集中。

解决方案是：rowkey 必须是 binary，且由上游（maxwell）按固定规则编码：

• BIGINT→Bytes.toBytes(long)→ 8 字节二进制；
• VARCHAR(32)→MD5(string).substring(0,8)→ 8 字节二进制；
• 复合主键 →Bytes.add(Bytes.toBytes(id), Bytes.toBytes(ts))。

于是 thrift 改为：

struct HBasePutRequest { 1: required binary rowkey; // 正确！ 2: required string table_name; 3: required list<ColumnValue> columns; } struct ColumnValue { 1: required string family; // cf 2: required string qualifier; // cq 3: required binary value; // 二进制值，由下游解析 4: optional i64 timestamp; // 可选，用于 HBase 版本控制 }

这个改动带来两个收益：

• HBase region 分布均匀：二进制 rowkey 的哈希更随机；
• 类型安全前置：maxwell 必须在发 Kafka 前完成 rowkey 编码，否则 thrift 序列化失败，问题暴露在链路最前端。

4.2 从 MaxwellEvent 到 HBasePut：5 步不可跳过的转换逻辑

thrift 层的 Java 服务收到MaxwellEvent后，并非直接put，而是执行以下 5 步：

1. Rowkey 构造：

   byte[] rowkey = RowKeyBuilder.build(event.getDatabase(), event.getTable(), event.getData().get("id"));

   RowKeyBuilder是一个策略类，按表名路由到不同编码规则（如user_db.user_profile用 MD5，order_db.order_header用 Long 编码）。

2. CF/CQ 映射：
   读取配置中心的hbase-mapping.json：

   { "user_db.user_profile": { "cf": "info", "mapping": { "id": "id", "name": "name", "email": "contact:email", "updated_at": "meta:updated_ts" } } }

   注意email映射到contact:email，即contact列族下的email限定符。

3. 空值与默认值注入：

   if (data.get("status") == null) { data.put("status", "active"); // 注入默认值，避免 HBase 存 null }

4. 敏感字段脱敏：

   if ("user_db.user_profile".equals(event.getDatabase() + "." + event.getTable())) { data.put("phone", Desensitizer.mobile(data.get("phone"))); // 138****1234 }

5. 构建 Put 对象：

   Put put = new Put(rowkey); for (Map.Entry<String, String> entry : data.entrySet()) { String[] parts = entry.getKey().split(":"); String cf = parts[0]; String cq = parts.length > 1 ? parts[1] : entry.getKey(); put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cq), Bytes.toBytes(entry.getValue())); }

这 5 步全部在 thrift 服务内完成，对外只暴露一个put(HBasePutRequest)接口。这意味着：业务逻辑变更（如新增脱敏规则）只需升级 thrift 服务，无需动 maxwell 或 Kafka。

4.3 Thrift 服务的部署模式：为什么我们不用嵌入式 Server，而坚持独立进程？

thrift 官方推荐用TThreadPoolServer嵌入到 HBase Client 中。但我们选择部署独立的thrift-hbase-writer进程，监听 9090 端口，Kafka 消费者调用其 RPC。

原因有三：

• 故障隔离：HBase 集群抖动时，thrift 服务可快速熔断，Kafka 消费者重试，不影响 maxwell 和 Kafka；
• 弹性扩缩：大促时可水平扩展 thrift 服务实例（如从 3 台扩到 12 台），Kafka 分区数不变；
• 监控统一：所有 HBase 写入请求都经过 thrift，可集中统计 QPS、P99 延迟、失败率。

部署拓扑如下：

Kafka Consumer (Java) ↓ (RPC over TCP) thrift-hbase-writer (3 instances, load balanced by Nginx) ↓ (HBase Java Client) HBase Cluster (12 Regionservers)

Nginx 做 TCP 层负载均衡，配置proxy_timeout 60s，避免长连接超时断开。

4.4 你必须埋的 3 个监控指标：它们比 HBase 的 metrics 更早预警问题

thrift 服务自身暴露的 metrics，比 HBase 的 JMX 更能提前发现问题：

其中rowkey_skew_ratio的计算逻辑是：
max(各实例处理的 rowkey 哈希值 % 100 的桶数量) / min(各实例处理的 rowkey 哈希值 % 100 的桶数量)
比值 >3.0，说明 rowkey 分布不均，需检查 RowKeyBuilder 规则。

5. HBase 写入不是 CRUD，它是对存储引擎特性的“逆向工程”

把 HBase 当作“分布式 MySQL”来用，是导致性能崩溃的根源。在这条链路里，HBase 写入层必须深度理解 HBase 的底层机制：MemStore 刷盘时机、BlockCache 命中率、Compaction 策略、以及 Region 分裂的触发条件。我们不是在“写数据”，而是在“调优存储引擎”。

5.1 MemStore 的隐形杀手：为什么putBatch大小必须 ≤ 200？

HBase 的Table.put(List<Put>)看似是批量操作，实则内部是循环调用单条put。每条Put会：

• 计算 rowkey 的 hash，定位到对应 Region；
• 将数据写入该 Region 对应的 MemStore（内存缓冲区）；
• 当 MemStore 达到hbase.hregion.memstore.flush.size（默认 128MB）时，触发刷盘（flush）到 HFile。

问题来了：如果putBatch一次传 1000 条，而这些 rowkey 全落在同一个 Region（比如全是user_id=1001的订单），那么这 1000 条会全部写入同一个 MemStore。一旦触发 flush，128MB 的 HFile 会瞬间生成，RegionServer 的 IO 和 GC 压力暴增。

我们的压测结论：putBatch大小必须 ≤ 200，且必须确保这批 Put 的 rowkey 尽可能分散在不同 Region。

实现方式是：在 thrift 服务中，对List<Put>做二次分组：

Map<String, List<Put>> groupedByRegion = new HashMap<>(); for (Put put : puts) { String regionName = getRegionName(put.getRow()); // 根据 rowkey 计算所属 region groupedByRegion.computeIfAbsent(regionName, k -> new ArrayList<>()).add(put); } // 对每个 region 分组，单独调用 put for (List<Put> regionPuts : groupedByRegion.values()) { if (regionPuts.size() > 200) { // 拆分成多个 batch for (List<Put> batch : Lists.partition(regionPuts, 200)) { table.put(batch); } } else { table.put(regionPuts); } }

经验：getRegionName()方法必须缓存 region 分配信息（从 HBase Master 获取），避免每次调用都走网络。我们用 Guava Cache，expireAfterWrite 10m。

5.2 BlockCache 的真相：为什么我们禁用 L1，只用 L2（Offheap）？

HBase 默认开启两级缓存：L1（on-heap，存放索引）和 L2（off-heap，存放数据块）。但我们的线上集群发现：L1 缓存命中率常年低于 5%，而 GC 时间飙升。

原因在于：maxwell 同步的数据，90% 是随机读（如风控查某个 user_id 的最新状态），而 L1 缓存的索引块（Index Block）是按 HFile 顺序加载的，随机读无法利用局部性原理。

解决方案：在hbase-site.xml中关闭 L1：

<property> <name>hfile.block.cache.size</name> <value>0.4</value> <!-- L2 占用 40% 堆外内存 --> </property> <property> <name>hbase.regionserver.global.memstore.size</name> <value>0.4</value> <!-- MemStore 占用 40% 堆内存 --> </property> <!-- 彻底禁用 L1 --> <property> <name>hbase.lru.blockcache.size</name> <value>0</value> </property>

效果：RegionServer Full GC 频率下降 70%，P99 延迟从 120ms 降至 45ms。

5.3 Compaction 的定时炸弹：为什么我们把 Major Compaction 改为每周日凌晨 2 点？

Minor Compaction（小合并）会自动触发，合并少量小 HFile；Major Compaction（大合并）会重写整个 Region 的所有 HFile，IO 压力巨大。

HBase 默认 Major Compaction 周期是 7 天，但起始时间是随机的。我们曾遇到：3 台 RegionServer 的 Major Compaction 在同一分钟内触发，集群 IO util 100%，所有写入超时。

解决方法：强制所有 RegionServer 的 Major Compaction 在同一时间窗口执行：

# 在每台 RegionServer 上执行 hbase org.apache.hadoop.hbase.util.HBaseFsck -fixReferenceFiles -fixHdfsOrphans # 然后设置 cron 0 2 * * 0 hbase org.apache.hadoop.hbase.util.HBaseFsck -fixReferenceFiles -fixHdfsOrphans

更稳妥的做法是：在 HBase Shell 中执行：

major_compact 'user_db:user_profile', '20240601020000' # 指定时间戳

提示：Major Compaction 期间，Region 仍可读写，但性能下降 30%-50%。务必避开业务高峰。

5.4 Region 分裂的临界点：如何用预分区让写入吞吐翻倍？

HBase 默认按hbase.hregion.max.filesize（默认 10GB）自动分裂 Region。但自动分裂是阻塞操作，分裂期间 Region 不可用。

我们的做法：建表时预分区（pre-split），按业务预期的 rowkey 分布，一次性划分足够多的 Region。

例如user_db.user_profile表，rowkey 是 MD5(user_id)，我们预分区 128 个：

create 'user_db:user_profile', {NAME => 'info', TTL => 2592000}, # 30 天 {NAME => 'contact'}, {NAME => 'meta'}, {SPLITS => [ '\x00\x00\x00\x00', '\x00\x00\x00\x40', '\x00\x00\x00\x80', '\x00\x00\x00\xC0', '\x00\x00\x01\x00', ... # 总共 128 个 split point ]}

128 个 Region 意味着：

• 写入请求可并行写入 128 个 MemStore，吞吐提升 10 倍以上；
• 自动分裂概率趋近于 0，稳定性大幅提升。

经验：预分区数 = 预期峰值 QPS / 单 Region 吞吐（我们实测单 Region 稳定吞吐为 800 QPS）。800 * 128 = 102400 QPS，远超当前 5000 QPS，留足余量。

6. 最终一致性不是妥协，而是用可验证的机制替代“相信”

整条链路的目标从来不是“强一致性”——那在分布式系统里是伪命题。我们的目标是：当 MySQL 主库发生变更后，HBase 中的数据在可接受的时间窗口内（≤ 3 秒），达到与源库逻辑等价的状态，且该状态可被任意第三方工具验证。

6.1 一致性验证的三板斧：checksum、sample、audit-log

我们每天凌晨执行一致性校验，用三套独立机制交叉验证：

1. Checksum 校验（全量，慢但准）：
   对 MySQL 和 HBase 中同一批user_id，分别计算MD5(CONCAT(id,name,email,updated_at))，比对 checksum 值。
   工具：自研hbase-mysql-checksum，用 MapReduce 扫描 HBase，JDBC 扫描 MySQL。

2. Sample 抽样（增量，快但覆盖有限）：
   从 Kafka 中随机抽取 1000 条update事件，检查 HBase 中对应 rowkey 的info:name和info:email是否与事件中data.name、data.email一致。
   工具：Python 脚本，5 分钟内完成。

3. Audit-log 回溯（定向，用于故障复盘）：
   在 thrift 服务中，对每条HBasePutRequest记录 audit log：

   { "request_id": "req-123456", "event_ts": 1718467200, "rowkey_hex": "a1b2c3...", "hbase_result": "success", "hbase_timestamp": 1718467200123 }

   当发现不一致时，直接查 audit log，定位是哪条事件写入失败。

6.2 断点续传的终极方案：基于 binlog_position 的幂等写入

Kafka 消费者commitSync()保证了 at-least-once，但 HBase 写入失败后重试，可能导致同一条事件写入多次。我们用binlog_position实现幂等：

• 在 HBase 中建一张hbase_audit表，rowkey = binlog_file:binlog_position；
• 每次写入