一个多月前，我们内部的数据平台需要把线上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:password@host1: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，如果只想保留整体，就用$.items，Kettle会输出一个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用的rowkey，rowkey_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或16MB，Kettle步骤执行时线程数设为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设计不合理，只能停掉重来，白白浪费时间。数据迁移这件事，慢就是快，前面多想一点，后面省力十倍。
一个多月前，我们内部的数据平台需要把线上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:password@host1: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，如果只想保留整体，就用$.items，Kettle会输出一个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用的rowkey，rowkey_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或16MB，Kettle步骤执行时线程数设为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设计不合理，只能停掉重来，白白浪费时间。数据迁移这件事，慢就是快，前面多想一点，后面省力十倍。