一个多月前我们内部的数据平台需要把线上MongoDB里的用户行为明细同步到HBase。下游要做多维统计和特征加工MongoDB的灵活文档模型开发期确实舒服但大范围scan和列式存储场景并不合适HBase能扛住高吞吐的随机读写与批量扫描正好接这个活。团队里有人主张写Java任务有人推荐Spark我评估一圈后选了Kettle也就是PDI。理由很简单这批数据涉及大量字段梳理、嵌套结构和频繁的临时补数纯代码方案每改一个字段映射都要走一遍编译、打包、发布流程而Kettle可以一边查Mongo一边调映射实时看到每行数据的变化调试效率高太多。这篇文章把这次迁移的完整思路和实际操作串一遍为什么选Kettle、迁移前字段怎么定、MongoDB和HBase两端怎么连接、转换流程怎么搭、增量同步怎么做以及我在真实环境里踩过的几个典型坑。如果你正在做或者准备做MongoDB到HBase的数据迁移下面这些内容应该能帮你少走不少弯路。1. 为什么在MongoDB到HBase的迁移里我坚持用Kettle先谈选型再看量级1.1 三条路线的对比当时我们内部其实讨论过三种主流方案自己写Java任务、用Spark集群跑批、用Kettle可视化ETL。我把它们拉了一张表从团队维护成本和临时同步的频繁程度上做了对比。Java任务最灵活什么字段都能处理但每个解析规则的变化都要改代码、打包、发布。如果对MongoDB游标、连接池不熟写出来的默认实现很容易在数据量上来之后暴雷。而且我们团队里并非所有人都是Java出身后续维护门槛偏高。Spark作业适合超大表的批量刷新性能上限也高。但Spark依赖的HBase Connector、版本兼容、集群资源申请一套流程对临时同步来说太重了。一个“查一下某天数据有没有少”的需求要经历改代码、提交任务、等调度实在不划算。Kettle ETL读取-转换-写出这个模型天然贴合迁移场景。字段映射直接在界面里拖数据流里的每一行都能预览出问题当场就能看到是哪一步的数据不对。我的结论很直接如果是超大规模一次性搬迁优先考虑Spark或HBase自带的bulk load工具链如果是持续在线业务、频繁字段调整、以及各种临时数据修复场景Kettle是效率和可维护性最好的选择。1.2 Kettle适合的数据量级边界Kettle不是银弹它有自己的量级边界。以我这次迁移的MongoDB集合为例大概有数千万条文档、每日新增百万级Kettle配合合理的批量参数完全能跑。但如果单集合到了十亿行以上还要一次性全量搬迁我建议还是老老实实走Spark或者HBase的离线导入体系Kettle更适合做“小步快跑”的增量同步和字段预处理。我个人经验大致是这样的千万行以内Kettle直接全量 增量配置好批量写入参数稳定可靠。千万到亿行可以跑全量但必须做好分批读取、游标控制和内存规划一次性塞进内存肯定炸。十亿行以上别硬顶。Kettle做数据抽样、格式清洗可以真正搬迁交给MapReduce风格的批处理或HFile方式导入。量级判断决定了后面所有设计所以我把这一点放在最前面说。如果你拿不准自己数据有多大先在MongoDB端跑一次count再决定是否继续往下看。2. 动手前必须定下来的模型设计rowkey、列族和Mongo文档的“翻译”规则2.1 先纠一个观念HBase不是把JSON塞进去就完事很多刚接触HBase的人有个错觉既然HBase的cell存的是字节数组那MongoDB的一条文档是不是转成JSON字符串塞进去就行我在最开始也差点这么干。实际上HBase的存储模型是“表-行-列族-列限定符-版本”rowkey和列族在设计阶段就必须确定查询按rowkey前缀和列范围来加速。换句话说MongoDB是文档驱动、字段可以随时变的数据库HBase是 schema 明确、靠rowkey和列族组织数据的列式数据库。迁移的关键不是“把数据倒过去”而是“按查询方式重新建模”。设计原则很简单未来想怎么查就怎么设计rowkey和列。先想清楚下游是哪些任务按什么条件取数再回头设计映射关系否则迁移完大概率还是全表扫描HBase的优势一点发挥不出来。2.2 rowkey设计从_id派生还是用业务键拼接rowkey是HBase里最高频的查询路径设计得好不好直接决定读写效率。我这次迁移的MongoDB集合里每条文档的_id本身就是ObjectId它是全库唯一的而且前4个字节是秒级时间戳。直接用它做rowkey有天然优势写入时按时间顺序插入读取时也能按时间范围做scan。但有一个经典问题如果直接使用原始id做rowkey写入新数据时可能集中在某些region上导致热点。网上常见的思路是加盐比如在rowkey前面拼一个hash(id) % 分区数的前缀。这样写入会分散到多个region但因为破坏了原始id的前缀一致性按原始id做范围扫描时就麻烦了。我的取舍很简单这次迁移的HBase表主要是下游离线任务按天扫描写入热点的压力远小于按时间顺序扫描的需求所以我保留了原始ObjectId字符串做rowkey只增加了少量随机前缀来缓解热写。如果你们场景是高频点查某一条记录比如按用户ID查那就应该直接把业务ID作为rowkey前缀不要留恋ObjectId。2.3 嵌套文档、数组字段怎么映射MongoDB的文档经常是这样的{ _id: ObjectId(64f7c2a1e4b0c3f5a1d2e3a4), event_type: click, user: { id: u_1001, region: east }, items: [sku_1, sku_2], created_at: ISODate(2023-11-09T12:30:00Z) }而HBase侧我需要明确每一列放在哪个列族下。我当时定义了一个列族cf_info把一条文档的字段翻译成一张映射表MongoDB字段HBase列转换说明_idrowkey的一部分转成十六进制字符串可能加盐前缀event_typecf_info:event_type字符串直接存user.idcf_info:user_id嵌套字段用下划线或点号拼成列名user.regioncf_info:user_region同上itemscf_info:items数组整体转JSON字符串或按元素拆行created_atcf_info:created_at建议存毫秒时间戳或UTC ISO字符串嵌套字段的处理比较直观用点号路径把user.id变成列名user_id。数组字段则有两个选择如果下游只是记录详情不需要按items里的单个元素做聚合直接存JSON字符串最省事如果下游要按某个item维度统计那就要在Kettle里把数组拆成多行每个元素一行同时把父文档的相关字段复制到每一行。数组拆行需要特别留意数据量放大问题——一条带20个items的文档拆完后HBase里会变成20行映射设计阶段一定要提前评估。2.4 版本策略别把HBase的版本能力用成存储黑洞HBase每个cell可以保存多个版本数据默认只保留1个版本可以设置成保留3个、5个甚至更多。我见过有人觉得“既然HBase天然支持多版本那把MongoDB每次更新都存进来以后能看历史变化”。听起来很美但版本越多占用的存储就越大。尤其是高写入频率的场景多版本会让region size膨胀得很快影响读写性能。我这次的同步策略是增量覆盖MongoDB源表里的文档更新后HBase覆盖写入同一rowkey、同一列的新值。HBase侧保持默认版本数1不保留历史。如果确实需要历史版本比如审计类数据再单独建一张历史表把变更记录按时间追加进去而不是依赖HBase的cell版本。3. Kettle两端的连接配置MongoDB好说HBase才是重头戏3.1 环境准备与驱动Jar别把Jar包扔错地方Kettle版本我用的社区版9.x。有个很关键的细节MongoDB相关步骤在不同发行版里的插件情况不一样老版本里叫“MongoDB Input”新版叫“MongoDB Document Input”。如果打开Kettle后步骤面板里搜不到MongoDB步骤多半是插件缺失需要先安装对应版本的MongoDB插件或者把MongoDB驱动jar放进lib目录。MongoDB驱动方面我建议直接放官方驱动包注意和服务端版本匹配。现代MongoDB驱动是mongodb-driver-sync这类模块化管理Kettle里可能只需要核心驱动和BSON库。把jar放进Kettle的lib目录后重启Spoon才能生效。HBase侧才是真正的重头戏。Kettle社区版自带对Hadoop/HBase的原生步骤但这些步骤的配置非常“随缘”经常出现类冲突或找不到类的问题。所以我后来采用的是自己写一个输出插件把HBase客户端相关jar全部隔离到插件的lib目录里而不是丢进Kettle主lib。这样Kettle启动时不会因为多个HBase版本混在一起而出怪问题。3.2 MongoDB连接配置连接串、认证、查询条件在转换里拖入一个“MongoDB Document Input”步骤双击打开新建连接。我使用的是MongoDB连接串方式mongodb://user:passwordhost1:27017,host2:27017/admin填写完连接串后点“Get Databases”能验证连通性能列出数据库列表基本就说明驱动和认证没问题。接下来选择数据库、集合然后进入字段和查询配置。这里提醒一个容易卡住的点Kettle的查询条件框里写的是JSON格式的Mongo查询语句不是Mongo Shell那种宽松语法。比如按时间区间查询直接写ISODate(...)是不行的需要用扩展JSON的$oid形式来构造ObjectId。后面增量同步时我会给出具体例子。字段输出方面可以在Kettle里配具体的字段路径比如$._id、$.event_type、$.user.id这样步骤会直接输出这些解析后的字段省掉后续一部分工作。但嵌套复杂时我更喜欢先输出整条JSON在后续的JSON Input步骤里做解析这样更灵活。3.3 HBase侧原生输出步骤、Phoenix和自定义插件怎么选Kettle自带HBase原生输出步骤但社区版配置体验并不友好。我当时试过“HBase Output”步骤要手动维护HBase连接信息、列族映射而且依赖Hadoop插件一旦类冲突就是各种NoSuchMethodError。所以实操上更推荐两条路线。路线A用Phoenix Table Output。如果集群里已经部署了Phoenix这是最快落地的方式。先在Phoenix里用SQL建好HBase表然后在Kettle里新建一个Generic Database连接连接类型: Generic database 驱动类: org.apache.phoenix.jdbc.PhoenixDriver URL: jdbc:phoenix:zk1,zk2,zk3:2181:/hbase之后拖一个“Table Output”步骤就能以标准的JDBC方式写入Phoenix底层由Phoenix完成对HBase的Upsert。优点是零Java代码缺点是多了一层Phoenix协处理器开销大吞吐写入时性能不如原生Put。路线B自定义Kettle输出插件直接调HBase客户端。这是我在生产环境里最终采用的方式。插件内部用HBase的BufferedMutator批量提交吞吐高、可控制性强而且完全绕开了Kettle原生Hadoop插件的那些坑。如果你团队里有Java基础我强烈建议走这一条。后面我会把插件里最核心的写入逻辑展开讲。4. 核心转换过程全拆解从嵌套JSON到批量Put的每一步4.1 一个完整转换的步骤编排我最终的Kettle转换是这样的步骤链步骤名类型职责Mongo ReaderMongoDB Document Input按条件读取MongoDB文档输出原始JSON字段Json ParseJSON Input解析嵌套JSON把user.id、user.region等路径拆成独立列RowKey BuilderModified Java Script Value拼rowkey、处理ObjectId、把日期规整成时间戳字符串Fix TypesSelect Values统一类型、处理null、去掉多余字段HBase Sink自定义输出插件写入HBase批量Put并flush步骤不多但每一步都有用途。第一个步骤只负责拉数据尽量不解析解析放到JSON Input里是为了避免MongoDB步骤直接解析时字段路径写错还要重新拉一遍数据。4.2 用JSON Input把嵌套文档拆成独立字段“JSON Input”步骤对熟悉Kettle的人来说应该不陌生。在上一步Mongo Reader输出一个名为raw_json的字段后JSON Input里配置这些JSONPath路径$.event_type $.user.id $.user.region $.items $.created_at$.event_type这种简单路径直接取字段值$.user.id则会把嵌套对象里的id取出来输出成独立字段。对数组items如果只想保留整体就用$.itemsKettle会输出一个JSON数组字符串如果需要拆行就在JSON Input里把items路径配置成重复元素路径每个数组元素都会交叉输出一行。这里有个小经验拆行前一定要想清楚目标行数。我刚开始为了下游统计方便把items数组拆成了多行结果统计行数发现从几千万条直接膨胀到几亿行把下游吓了一跳。后来改成整体存JSON字符串下游要分析时再单独解析效果反而好。4.3 用JavaScript步骤做rowkey和字段规整JSON Input拆出来的字段还需要再过一遍“Modified Java Script Value”步骤主要做三件事拼rowkey、格式化时间、统一字段类型。我当时脚本的核心逻辑是这样的// 拿到ObjectId的字符串形式Mongo步骤通常已转成十六进制字符串 var oid _id; // 简单加个盐前缀缓解写入热点问题前缀范围可配置 var salt 0; try { salt Math.abs(oid.hashCode()) % 4; } catch(e) { salt 1; } var rowkey salt _ oid; // 日期字段统一转成UTC毫秒字符串避免后续时区问题 var createdMillis new Date(created_at).getTime(); if (isNaN(createdMillis)) { createdMillis 0; } var rowkey_date createdMillis.toString(10);rowkey字段是给HBase用的rowkeyrowkey_date可以作为列存入cf_info:created_at。这里要注意HBase的cell值本质是字节数组如果你希望下游按数值范围扫描就别存“2023-11-09 12:30:00”这种人类可读字符串直接存毫秒时间戳字符串会更利于排序和过滤。存字符串还有一个好处写完HBase后再查出来至少不会因为字节序导致排序错乱。类型统一也在这步做。比如从JSON里读出来的数字型字段如果目标是让Phoenix做SQL聚合就必须保证和Phoenix表里定义的类型匹配如果是原生HBase Scan我建议统一转成字符串因为字典序在绝大多数情况下更可控。4.4 批量写入参数与失败重试的工程化处理如果是自定义插件写入逻辑我用的是HBase的BufferedMutator。这个类相当于一个缓冲区多个Put先攒在内存里达到阈值后异步批量提交比一个一个Put快很多。插件里核心代码思路大概是BufferedMutatorParams params new BufferedMutatorParams(TableName.valueOf(your_table)) .writeBufferSize(8 * 1024 * 1024); // 8MB缓冲 params.listener((Exception e, Row[] rows) - { // 这里接收写失败的批次记日志后续补写 }); BufferedMutator mutator connection.getBufferedMutator(params); // Kettle的processRow方法里每解析好一行就构建一个Put Put put new Put(Bytes.toBytes(rowkey)); put.addColumn(Bytes.toBytes(cf_info), Bytes.toBytes(event_type), Bytes.toBytes(event_type)); mutator.mutate(put); // 在插件dispose方法里统一flush并关闭连接 mutator.flush(); mutator.close();几个参数我调过很多次最终稳定下来的配置是writeBufferSize给到8MB或16MBKettle步骤执行时线程数设为4到8。写入线程数不是越大越好。我第一次开16个线程并发写直接把HBase集群写满了region频繁分裂反而触发一堆写入超时。后来降到6个线程配合8MB缓冲整体写入速度反而更稳。失败重试我是在listener里处理的。BufferedMutator本身会重试一部分可重试的异常但像RetriesExhaustedWithDetailsException这种就要把失败的行收集起来写到本地的补数表里等晚上低峰期再跑一次同步。千万不要在listener里直接无限重试集群一旦抖动越重试压力越大容易形成雪崩。5. 增量同步怎么实现用ObjectId自带的时间戳做一个低成本水位5.1 ObjectId时间戳的增量原理MongoDB的ObjectId是一个12字节的值前4字节是时间戳单位是秒。这个特性让增量同步变得非常简单我们不需要在MongoDB集合里额外维护一个“更新时间”字段直接用_id就能构造时间区间。思路是记录上次同步到的时间戳比如last_ts然后把它转成十六进制前面补0到8位后面分别拼上最小值和最大值构造出这一个时间点对应的ObjectId区间最小ObjectId: 十六进制时间戳 0000000000000000 最大ObjectId: 十六进制时间戳 ffffffffffffffff然后在Kettle里对MongoDB的查询条件写成{ _id: { $gte: { $oid: 64f765400000000000000000 }, $lt: { $oid: 64f86b1dffffffffffffffff } } }这个查询会走_id索引不必全表扫描效率很高。第一次全量迁移时把last_ts设成0$gte的对象就是最小值等同于全扫。5.2 Kettle作业编排全量、增量与状态记录我在Kettle里用Job串联流程是这样的先执行一个SQL步骤从状态表sync_checkpoint里读上一次同步的时间戳last_ts写入Kettle变量。执行同步转换。转换里的MongoDB查询条件用变量拼出最小/最大ObjectId。转换结束后再执行一个SQL步骤把当前批次的最大时间戳更新回sync_checkpoint表。状态表结构很简单CREATE TABLE sync_checkpoint ( target_table VARCHAR(100) PRIMARY KEY, last_ts BIGINT NOT NULL, update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP );这里有个小问题要注意ObjectId时间戳是秒级精度同一秒内写入的多条数据如果上一批读到这一秒的中间位置下一批可能会漏掉同秒但尚未被读取的文档。为了稳妥我在实际实现里会把last_ts往回退几秒也就是上一批处理到T秒下一批从T - 5秒开始重新扫这一部分数据。HBase是覆盖写同一rowkey重复写不会有重复数据所以“多读一批”比“漏读一批”安全得多。5.3 校验与补数别让“看起来同步成功”骗了你增量同步跑完一定要校验。我在每个批次结束后会做两个动作第一用MongoDB的countDocuments统计符合时间区间的文档数和HBase侧实际写入的行数做对比第二随机抽几个rowkey回查MongoDB原始文档比对HBase里的对应列值。如果两边数量对不上我会把sync_checkpoint的时间戳直接回拨到出问题的位置附近再用同一套转换重跑。因为是覆盖写重复执行不会产生脏数据。这个补数思路在业务上非常实用尤其是那些“不知道哪批数据丢了但必须补回来”的场景。6. 真实迁移中我踩过的坑类冲突、游标超时、写入抖动与时区错位6.1 类冲突HBase客户端Jar与Kettle自带Hadoop的“暗战”这是一个让我折腾了一整天的坑。现象是自定义插件一启动马上报NoSuchMethodError或者ClassNotFoundException类名是org/apache/hadoop/hbase/client/Connection之类的。第一反应是缺包但把HBase客户端jar全塞进Kettle的lib后反而报了更多冲突。后来排查发现Kettle本身自带了一部分Hadoop相关依赖版本和HBase集群的客户端版本不一致。两个不同版本的类同时出现在classpath里谁先加载完全看运气结果就是各种诡异错误。我的解决方案是自定义插件不要依赖Kettle主lib里的HBase包把所有HBase客户端依赖打进插件的独立jar包里并利用PDI插件机制自带的classloader隔离让插件运行时优先加载自己的HBase版本。这样Kettle主程序用它的Hadoop插件用自己的HBase客户端互不干扰。如果你不是用自定义插件而是直接往Kettle主lib里塞HBase驱动一旦遇到类似问题先对比一下Kettle自带hadoop相关jar版本和集群版本尽量做到一致。6.2 MongoDB游标超时和Kettle内存溢出第一次跑千万级全量时跑到一半MongoDB报游标cursor not found同时Kettle界面内存占用直线上升最后直接卡死。这两个问题本质上是相关的MongoDB服务端的游标默认10分钟没有操作就会过期而Kettle默认给Spoon分配的内存又比较小如果Mongo Input一次性拉回大量文档积压在内存里很快就会OOM。我的解决办法有两步。第一步是修改Spoon的启动脚本把堆内存调大比如-Xmx4g或-Xmx8g同时关闭Spoon的自动刷新预览避免界面刷新占用大量CPU。第二步是把大的全量查询拆成小段。具体做法就是前面增量同步里设计的ObjectId区间查询不再用一个长游标一次拉完而是按小时或按天切段每段拉完立即写入HBase并释放内存。这样虽然多跑几轮循环但内存稳定也不会触发游标过期。6.3 写入高峰期HBase Region Split导致的连接中断这个坑发生在白天迁数正好跟线上其他任务抢HBase资源。现象是转换跑到一半突然大面积报IOException错误信息里有Call to xxx timed out。刚开始我怀疑是网络问题但ping和telnet都正常。后来观察HBase集群监控发现那段时间经常触发region split加上GC停顿导致RPC请求超时。应对方案有几个我最终组合使用了降低Kettle转换的并行线程数从16降到6减少同一时刻写入压力。把客户端写缓冲调大让数据在本地攒够一批再提交降低RPC频率。错峰运行把日常增量同步放在凌晨低峰期避开其他任务的高峰。如果你们HBase表还没设计预分区写热点也会加剧这类问题。rowkey加盐是简单有效的缓解手段这也是我在第2章强调rowkey设计的原因。6.4 时区错位与JavaScript解析日期的连环坑迁移完成后下游跑统计任务发现部分日期差了8个小时。排查下来是两个环节叠加造成的。第一Kettle在读MongoDB的日期型字段时驱动默认会转成本地时区显示如果你在Kettle预览里看到的时间是“2023-11-09 20:30:00”但MongoDB里存的实际是UTC的“2023-11-09 12:30:00”那后面所有基于这个值的处理都会带着时区偏移。第二JavaScript步骤里用new Date(created_at)解析字符串时不同版本的解析规则不一样有的能识别ISO字符串有的则返回NaN导致时间字段变成0或空值。我的规整方式是所有日期字段在离开MongoDB之后立刻统一转成UTC毫秒时间戳字符串不做本地时区格式化也不要试图用人类可读字符串传参。到HBase里存的就是毫秒时间戳下游解析时自行决定时区显示这样两边永远不会因为时区差8小时扯皮。最后说几句这套MongoDB到HBase的迁移方案在我们内部已经稳定跑了蛮久。Kettle的价值不在于它比Spark快而在于整个迁移过程可以边看边调一个字段映射不对拖一下步骤就能修正不用重新编译发布。对运维和临时补数来说这种灵活性的收益是实打实的。最后分享一个体会不管选哪条HBase写入路线先抽一万条数据试跑一次把字段映射、rowkey规则、批量参数都调试稳定了再放开跑全量。我见过太多人上来就全量同步结果跑到一半发现rowkey设计不合理只能停掉重来白白浪费时间。数据迁移这件事慢就是快前面多想一点后面省力十倍。
