Kettle实战：MongoDB到HBase数据迁移完整流程与调优

数据迁移这件事，最怕的不是数据量大，而是数据模型对不上。MongoDB里面存得挺规整的一条JSON文档，到了HBase这边要拆成一行行带rowkey的单元格，稍不留神就把嵌套字段搞丢了。最近帮一个团队做Kettle把MongoDB数据往HBase迁移的活儿，踩了不少坑，也理出了一套能直接用的流程。这篇就围绕这个场景，把从思路设计到实操细节、再到问题排查的完整过程写出来，给后面要接类似任务的同学做个参考。

1. 整体思路与迁移方案设计

1.1 迁移任务的核心矛盾：文档模型 vs 列族模型

MongoDB和HBase的存储逻辑差异很大，这是整个方案设计的出发点。MongoDB是文档型数据库，一条记录对应一个JSON文档，字段可以嵌套，数组随便塞，schema基本不受约束。HBase则是标准的列族宽表模型，一行数据由rowkey加若干个列族（Column Family）组成，每个列族下面再有列限定符（Column Qualifier），底层按rowkey字典序落盘。

所以，迁移工作本质上做的不是"搬数据"，而是"转换数据模型"。把MongoDB里嵌套的文档打平成HBase的列，把数组想办法拆行或者序列化，把_id或者某个业务字段提升为rowkey。这个转换逻辑想不清楚，后面的Kettle作业设计就是空中楼阁。

这次任务的数据背景是一个某业务系统的用户行为日志库，单集合大概几十亿条文档，字段有常规的字符串、时间类型，还有不少嵌套对象和数组。目标是把这些数据落到HBase的某张用户行为表中，供下游做离线分析和查询。

1.2 为什么选Kettle而不写Spark作业

有人可能会问，这么大规模的迁移，直接写个Spark或者MapReduce脚本不香吗？非要绕一圈用Kettle，图什么？

我的判断标准很简单：看迁移任务是一次性还是周期性。如果是纯一次性的大批量“搬库”，写Spark作业确实更高效。但这个项目的真实情况是，第一轮全量迁移之后，每天还有增量数据从上游MongoDB同步过来，而且MongoDB那边还不断有字段结构调整的需求。这种场景下，用Kettle的价值就体现出来了：

• Kettle的作业和转换都是图形化配置，字段映射、类型转换、过滤规则这些逻辑改起来非常快，不需要重新编译打包。
• 有现成的MongoDB输入插件和HBase输出插件，内置读写能力，不用自己写连接器。
• 日志、错误处理、调度这些能力都自带，接手的人上手成本低。

所以方案选型的核心原因是：需要持续维护和迭代的ETL任务，用Kettle这类可视化工具，长期ROI更高。

1.3 整体迁移架构与流程设计

整个迁移链路分为三层，串起来是这样的：

• 数据抽取层：Kettle通过MongoDB Input插件读取某个集合的数据，分页抽取，避免一次性加载全量导致内存溢出。
• 数据转换层：用Kettle的“增加序列”生成唯一的业务主键，然后用Java代码或者映射节点把嵌套的Document对象解析成扁平的字段。
• 数据加载层：通过HBase Output插件写入目标表，rowkey由业务主键转换而成，列族映射关系预先配置好。

作业设计上，我强烈建议把“读MongoDB”和“写HBase”拆成两个独立的转换，中间用作业里的“转换步骤”串联。这样有一方出问题时，不需要从头跑整个链路，可以只重启失败的环节。

2. 核心难点拆解与关键技术细节

2.1 MongoDB文档结构分析与字段映射策略

动手配置Kettle之前，务必要做的一件事是分析MongoDB里文档的实际结构。不要只信开发给的接口文档，实际数据里往往藏着很多意外，比如某条文档没有某个字段、嵌套层级深浅不一、数组元素类型不一致等。

先用脚本或者在MongoDB客户端里抽样几个文档，把层级结构理清楚。以我们这次的用户行为日志为例，典型的文档长这样：

{ "_id": ObjectId("5f2d1c0b2d3a4b5c6d7e8f90"), "user_id": "U12345", "event": "click", "timestamp": ISODate("2024-01-15T08:30:00Z"), "page": { "url": "/home", "title": "首页", "duration": 120 }, "tags": ["mobile", "china"], "extra": { "os": "iOS", "app_version": "3.2.1" } }

HBase这边的目标表设计为两个列族：info和detail。info存核心用户标识，detail存事件详情和扩展字段。

字段映射关系如下：

这里有一个很重要的经验：嵌套对象和数组，不要强行打平到多个列。虽然在HBase里你可以把page.title、page.url展成两个列，但是如果嵌套层级有三四层，或者字段数量是动态的，打平的成本就会指数上升。常规做法是：

• 嵌套深度固定且字段业务上需要单独查询的，才打平成独立列。
• 否则直接整个子文档序列化成JSON字符串，塞到一个列里，等查询时再解析。

2.2 rowkey设计：从MongoDB _id到HBase rowkey的转换

rowkey设计是HBase这侧最关键的决策点，直接决定写入时是否产生热点、查询时能否命中region。

很多人的第一反应是直接用MongoDB的ObjectId字符串作为rowkey，这在逻辑上没问题，但有一个隐患：如果业务大量写入的_id前缀相同（比如同一秒生成的ObjectId前缀一样），会导致写入全部打在同一个region上，造成数据倾斜。这个在数据量小的时候没感觉，一旦上了几十亿条，region分裂和热点写问题会非常头痛。

这次用的方案是：userId的hash值 + MongoDB _id的尾部字符拼接成组合rowkey。

// 伪代码示例，实际在Kettle的Java代码步骤中实现 String uid = "U12345"; String oid = "5f2d1c0b2d3a4b5c6d7e8f90"; int hash = Math.abs(uid.hashCode() % 100); // 根据uid哈希分散 String rowkey = String.format("%02d_%s_%s", hash, uid, oid.substring(oid.length() - 8));

这样做的直接好处是写入请求会均匀分散到多个region，同时查询时只要知道userId，可以快速算出前缀，定位到具体的rowkey范围，不会因为引入随机数而导致无法按前缀扫描。

2.3 字段类型映射表：MongoDB Type 到 HBase 的处理规则

MongoDB的数据类型比传统关系库丰富不少，比如有Date、ObjectId、Long、Array、Embedded Document等。这些类型在往HBase写的时候，大部分最终都落成字节数组（byte[]），但前置的转换逻辑要慎重点。我自己整理了一套映射规则，基本覆盖了常见情况：

这表看着简单，但有两个容易翻车的地方：一个是Date，如果不转long而直接ToString，后续做时间范围查询的时候就没法利用HBase的字典序比较，会很吃亏；另一个是Null值，HBase本身不存储null列的，写入时直接跳过就好了，否则会把一条空值列也占一个版本存储。

3. Kettle环境准备与插件配置精讲

3.1 开发环境与插件选型

做之前先确认Kettle版本。这个项目用的是Kettle 8.2版本，对应的MongoDB插件和HBase插件都是自带的，不需要额外安装。如果你用的版本比较新或者比较老，插件确认步骤建议看一眼：打开Kettle后，左侧“Big Data”或“NoSQL”分类下有没有MongoDB Input和HBase Output这两个步骤。没有的话，去对应市场装一下。

环境这边有个小坑：HBase插件依赖Hadoop相关jar包，Kettle自带的这些jar版本可能和集群不完全匹配。所以操作上，我建议把集群端的hbase-site.xml、core-site.xml、hdfs-site.xml拷贝到Kettle所在机器，通过classpath引入。这个看起来是杂活，但搞不定的话连接HBase时会报各种莫名其妙的RPC错误。

3.2 MongoDB输入插件配置细节

在Kettle里新建一个转换，拖入“MongoDB Input”步骤。配置里有几个重点项：

• Connection：填写MongoDB连接串，格式为mongodb://host:port/database。如果开启了认证，用户名密码也可以在高级选项里配上。
• Database：库名。这里要注意，Kettle的MongoDB插件有版本差异，有些版本需要额外指定认证库，否则认证失败。
• Collection：集合名，比如user_events。
• Lookup：这个对应的是查询条件。默认情况是空查询，也就是全量抽取。如果要做增量，可以在这里写{ "sync_time": { "$gte": ISODate("2024-01-15T00:00:00Z") } }。
• Projection：指定需要导出的字段。强烈建议在这里用projection只查业务需要的字段，别一股脑把不用的嵌套大字段也拉出来，浪费IO和内存。
• Limit，Skip：这两个是在Kettle端做的分页逻辑。全量几十亿条时不要直接设Limit很大，建议分批次，配合作业循环执行。

还有一个细节：MongoDB输入插件读取出来的字段类型，Kettle会做标记。比如嵌套文档会被标记为“Document”类型，数组会被标记为“Array”类型。这些类型在后续Java代码步骤里会以MongoDB的DBObject或者类似结构呈现，没法直接当普通字符串处理，需要针对性地写解析逻辑。

3.3 HBase输出插件配置与映射绑定

HBase输出侧，Kettle里会用到一个“HBase Output”步骤。配置项不少，但核心是这几块：

• HBase connection：需要提前在Kettle里配置HBase连接。这步要求填Zookeeper quorum地址、端口，以及读取hbase-site.xml。
• Target table：目标表名，比如user_event_log。如果表不存在，Kettle是可以自动建表的，不过建议手动预建，因为表的分区策略和预分区要自己控制。
• Mapping：这是核心，需要把Kettle的流字段绑定到HBase的列族和列限定符上。操作逻辑是：左侧选择流里的字段，右侧选择目标列（格式为 列族:列限定符），然后绑定。
• Row key：选择一个流字段作为rowkey。通常这里选的就是我们在转换里拼好的rowkey字段。
• Write buffer size：批量提交缓冲区大小，默认值比较保守，建议调大，比如1MB或2MB。写入吞吐量会有明显改善。

3.4 常用工具类：在Kettle中写代码解析文档

Kettle的图形化组件虽然方便，但碰上MongoDB的嵌套文档解析，还是得靠代码。这里推荐用“Java代码”步骤，它和“脚本”步骤的区别是执行时效率更高、调试更方便。

以一个典型场景为例：把MongoDB的嵌套文档page解析出来，并单独生成两个字段。

// 这里假设前面的字段名是"page"，类型为Document Object pageObj = get(Fields.In, "page").getValue(r); if (pageObj != null) { // 在Kettle的Java代码步骤中，嵌套文档通常是DBObject DBObject pageDoc = (DBObject) pageObj; String url = (String) pageDoc.get("url"); Object durationObj = pageDoc.get("duration"); putData("page_url", url); putData("page_duration", durationObj == null ? 0 : Integer.parseInt(durationObj.toString())); } else { putData("page_url", null); putData("page_duration", 0); }

有几个写代码时容易忽略的点：

• MongoDB的DBObject里通过get取出来的值，如果是嵌套层，往往是Object类型，不能直接强转成String或者Integer。要先toString再做类型转换。
• 如果数组里混了类型，比如一个数组里有字符串又有整数，序列化成JSON时不要自己拼字符串，推荐用Java里现成的JSON库，避免引号转义出错。
• 代码步骤的putData和get(Fields.In, ...)是Kettle推荐的字段操作方式，别用RowMeta手动操作，容易出错。

4. 实操过程：完成一次全量迁移的详细步骤

4.1 预建HBase表结构与预分区

先在HBase Shell里把目标表建好。这次的项目我们根据rowkey前缀做了预分区，前缀范围从00到99（对应hash到100），一共100个region。

create 'user_event_log', {NAME => 'info', VERSION => 1}, {NAME => 'detail', VERSION => 1}, {NUMREGIONS => 100, SPLITALGO => 'HexStringSplit'}

这里有一个设计决策要说明一下：因为我们rowkey的前缀是两位数字（00~99），所以用HexStringSplit做预分区，可以保证每个region只负责一小段前缀对应的rowkey。如果不做预分区，写入初期所有数据都会落在单个region上，随着region自动分裂才会有分散，但分裂本身也有成本，大批量写入高峰期容易把HBase Master拖垮。

4.2 编写全量抽取转换作业

Kettle里的作业是多个转换的载体，我们全量抽取时用作业串联两个转换：

转换一：读取MongoDB并解析

流程是：MongoDB Input → Java代码（解析嵌套字段）→ 增加序列（生成ID）→ 拼接rowkey → 输出到下一个转换需要的临时存储。

这里有一个我强烈推荐的做法：用"写文本文件"步骤把转换一的结果落盘到本地临时文件，而不是直接在内存中传给下一个转换。原因很简单：几十亿条数据全放Kettle的虚拟管道里，运行到后期内存会撑爆。落到文件里，转换二再逐行读取，虽然多一点磁盘IO，但稳定性提升很明显。

转换二：读取临时文件并写入HBase

流程是：文本文件输入 → 字段类型整理 → HBase Output。

这种设计还有一个好处：转换二单独运行时可以反复读写同一个临时文件，不影响源头MongoDB的连接。某个环节失败了，不必从头跑全量抽取。

4.3 增量迁移策略与去重方案

全量跑完之后，每天要接MongoDB里的增量数据。增量同步我一般用下面这套策略：

• MongoDB里的文档建议带上一个最后更新时间字段，比如sync_time。增量抽取时查询条件直接写{ "sync_time": { "$gte": 上次同步时间 } }。
• 但是MongoDB和HBase两边最终一致性很难靠时间戳保证，因为可能存在业务数据回刷，比如一条旧记录更新了但sync_time没变。
• 所以推荐一个更稳的方案：在HBase里把rowkey设计成业务唯一ID + 更新时间，例如hash_uid_timestamp。这样同一业务ID的多次更新，在HBase里会有多个版本，查询时取最新版本即可。HBase的版本控制天然支持这个机制，无需额外去重。

alter 'user_event_log', {NAME => 'info', VERSION => 5}

把info列族的VERSION从1改成5，意味着最多保留5个版本的单元格数据。上游同步过来的同一条uid的多次更新，全部落库，读取时通过设置时间范围或者版本号取最新值。

4.4 性能参数调优：Kettle、MongoDB、HBase三端配合

全量迁移过程中，性能瓶颈往往是连锁反应。我列举几个调优过并见效的参数：

Kettle侧：

• KETTLE_MAX_JOB_TRACKER_SIZE这个环境变量建议调大一点，否则作业日志记录多了会堆积。
• 转换里的“提交大小”（Commit Size）并不是越大越好，提交过大时HBase的批量flush如果失败，回滚成本很高。我一般设置在2000到5000条之间。
• 如果机器核数够，可以多开几个并行转换，每个转换负责读取不同范围的MongoDB数据，然后并写HBase。

MongoDB侧：

• 抽取时尽量走项目里配置好的MongoDB从节点，避免影响线上业务。
• 查询最好命中索引。如果按sync_time增量抽取，这个字段必须建有索引，否则全表扫描会把源库的CPU打满。

HBase侧：

• 写入时关闭WAL？不要随便关。WAL保证数据可靠性，正常情况下不建议为了性能关掉。如果确实测试环境，可以临时关。
• hbase.client.write.buffer调大一点，让单次批量提交包含的行数更多。
• 注意观察RegionServer侧的GC。如果频繁Full GC，说明写入太快导致MemStore过大，可以适当调低Kettle侧的写入并发。

4.5 数据校验：如何确认迁移没有缺漏

迁移完了，心里没底怎么办？数据校验是必须做的一环，不能只看“程序跑完没报错”。建议从三个维度交叉验证：

• 条数校验：用MongoDB的count和HBase的count做对比。但注意，HBase的count命令在大表上非常慢，建议用Hive关联HBase表来做count，或者用rowkey范围区间分别统计。
• 字段级抽查：选几条已知的业务ID，从MongoDB原库捞出来，再按rowkey从HBase里查询，逐字段比对type和时间格式。
• 汇总校验：如果业务允许，对某个数值型字段做sum操作，两边各算一遍做对比。比如日志表里的时长duration字段，全量表sum一次，误差应该在可接受范围内。

这里需要注意，如果是增量写入导致同一rowkey存在多个版本，count对上不能代表内容正确，还需要校验“当前版本值”（即最新时间戳对应的列值）是否和MongoDB一致。

5. 常见问题与排查技巧实录

5.1 读取MongoDB时内存溢出

现象：Kettle的MongoDB Input步骤跑了一段时间，JVM报OutOfMemoryError。排查过程：一开始以为是Kettle分配的堆内存不够，调大到8G还是报。最后定位到原因是MongoDB Input插件在抽取时会加载整个集合的一部分到内存，如果Kettle的“MongoDB Input”步骤没有正确分片和分包，聚合到内存的记录数会非常大。

解决办法：

• 在查询条件里增加$limit和$skip，手动把全量切分成多个子集，分批抽取。
• 确认是否使用了Projection来裁剪字段宽度。字段宽了，就算行数不多，内存消耗照样大。
• Kettle启动脚本里的JVM堆参数对应调大，比如-Xmx8g。

5.2 HBase写入时报RegionTooLargeException

现象：HBase在批量写入过程中报RegionTooLargeException或者RegionTooBusyException。排查过程：这个异常通常是因为region内的数据量超过了阈值，触发了自动分裂，但分裂期间的大量写入都打过来，导致RegionServer处置不过来。

解决办法：

• 提前做预分区，避免一边写一边分裂。
• 适当调大hbase.hregion.max.filesize，让region晚一点触发分裂。
• 把写入并发降低，别让RegionServer压力过大。

5.3 Kettle里的DBObject取出来的数字变成了Double

现象：从MongoDB读取到的某个整型字段，在Kettle的Java代码中打印出来变成1.0，导致写入HBase时数据类型不对。

排查过程：MongoDB返回的数值类型可能不止Int32，还有Int64和Double，DBObject把数字统一反序列化成了Double或Long。直接Integer.parseInt(durationObj.toString())在值为1.0时会报错。

解决办法：

• 不要直接转Integer，先判断类型，或者统一用Decimal转换。
• 推荐写法：

Object numObj = pageDoc.get("duration"); if (numObj instanceof Double) { duration = ((Double) numObj).intValue(); } else if (numObj instanceof Integer) { duration = (Integer) numObj; } else { duration = Integer.parseInt(numObj.toString()); }

5.4 映射配置了但HBase写不进去任何数据

现象：日志显示每一批处理了2000行，但HBase表里一条都没有。排查过程：一开始怀疑是HBase连接问题，检查连接正常。后来逐字段看HBase Output步骤输出的统计，发现rowkey字段是空的。再查发现是“拼接rowkey”的那个Java代码步骤没有正确输出字段。

解决办法：

• 检查Java代码步骤的“输出字段”里是否声明了包含rowkey在内的所有新字段。
• 在HBase Output步骤的“row key”配置里不要选"无"，要明确选择对应的流字段。

5.5 时间字段时区偏移8小时

现象：HBase里存的timestamp字段，读取出来比MongoDB里的原值少了8小时。

排查过程：Kettle的MongoDB插件在读取Date类型时，会按JVM的默认时区进行解析，而集群的JVM时区如果不是UTC，就会出现偏移。

解决办法：

• 在Kettle启动脚本的JVM参数里显式指定时区：-Duser.timezone=UTC。
• 或者在上游转换直接把时间转为long毫秒，根本不用Date类型去传递，这样彻底规避时区问题。

我个人每次做这类MongoDB到HBase的迁移，都会把时区、空值、字段类型这三件事当作前置检查项。这三个地方出问题的概率比想象中高得多，而且往往在你跑完几十亿条数据做校验时才暴露出来，回溯成本极高。所以流程上宁可前面多花一点时间把文档结构摸清楚，把映射规则定死，也不要中途再反复调整。

最后再分享一个小技巧：Kettle作业跑批量迁移时，一定要开详细日志，并且把日志定期切割保存。如果中途出了问题，全靠日志回溯。别指望在界面上盯进度，几十亿条数据的迁移时长远超你的耐心范围。日志里多打印每一步的行数和耗时，是排查性能瓶颈和定位错误数据的最直接手段。