★ FEATURED ARTICLE 简介面向KettlePDI开发与维护人员的一份实操说明文档聚焦循环获取结果集数据并传入转换处理的高频需求。文档以Job与两个转换t1.ktr、var.ktr为主线先说明t1.ktr生成结果集再通过JavaScript步骤中的previous_result.getRows()取出上一步数据配合parent_job.setVariable保存结果总数、循环控制变量i以及当前行的id、name字段随后在转换var.ktr中获取这些变量逐行输出到本地txt文件完成从取数、传参到落地的完整闭环。包体为单个PDF文件大小约130KB目前已有2365人学习/下载。内容包含j1.kjb中的完整代码参考、转换配置说明与最终输出结果验证并对循环中的变量更新、空结果判断等细节给出了具体写法。无论是日志表批量导出、按库表清单循环抽取还是将临时结果集拆分为多批次任务都可复用这套机制适合需要快速掌握Kettle跨转换传参和循环迭代处理的ETL开发人员直接对照使用。1. Kettle循环获取结果集中的数据并传入转换先解决“取一批、逐行跑”的批处理卡点如果你用Kettle做数据抽取大概率遇到过这种需求第一步从库里查出几十个ID第二步针对每个ID去更新数据、调用接口或跑子查询。直接连两个“表输入”做不到因为第二个处理动作需要的是第一个查询结果里的每一行而不是整张表。这个场景就是Kettle循环获取结果集中的数据并传入转换。它解决的问题是“先查一批、再逐行消费”的批处理任务特别适合定时跑批、按账号循环处理、维表增量同步。新手卡在结果集不生效熟手卡在变量传不进去。接下来按我实际排查的顺序讲透。2. 结果集在Kettle里怎么存复制行到结果与从结果获取记录的配合Kettle结果集是运行期内存中的行集合由“复制行到结果”写入由“从结果获取记录”读取。这两个步骤虽然名字直白但行为不像你想象的那样对称。先说“复制行到结果”它会把输入的所有行一次性复制到内存的结果集里而且不是移动是复制。源行仍然会在转换流里正常流动所以如果你在同一个转换里把它接在“表输入”后面既想保留原流、又想给结果集留一份是可以做到的。需要注意的是结果集不是数据库表更像一个队列读取即消费不能回头。2.1 用“复制行到结果”把数据写进内存结果集实际场景里“复制行到结果”通常放在一个转换的最后。这个转换可能是一条查询比如用“表输入”查出一批订单号经过“字段选择”保留order_id和cust_id最后连上“复制行到结果”。它不产生输出文件只把行集暂存在JVM内存里供后续作业项或转换使用。这个步骤的配置几乎为零没有连接属性没有高级选项。真正要花时间的是它前面的字段整形。我一般会在查询转换的SQL里给字段起英文别名避免数据库返回ORDER_ID这种大写或中文列名。因为结果集后面的步骤会严格按字段名取值大小写不一致最容易出问题。另外“复制行到结果”每次重新执行都会用当前输入覆盖掉上一次的结果集而不是追加。这个特性在循环里是好的至少不会把旧数据带进去但如果你的整体方案是“先查一次、后面循环里反复读”那它就会变成坑后面第5章会专门说。为了快速确认这一步有没有把数据写进结果集我习惯在“复制行到结果”前面加一个“行数统计”步骤把总数写到日志里。比如订单总数1286这样一旦后续循环没跑起来你能直接判断是查询没数据还是循环配置有问题。Kettle菜鸟教程很少提这个调试思路但实际跑批非常管用。2.2 “从结果获取记录”的读取顺序与游标特性“从结果获取记录”读取的是结果集里的行读取顺序和写入顺序一致。它有两条常年被忽略的行为只向前、读取即消费。你可以把它当作一个队列。如果你在同一个转换里连续放两个“从结果获取记录”第一个步骤把行取走之后第二个步骤什么都拿不到。这就是为什么很多人想在单个转换里实现循环最后翻车的原因。和数据库游标对比这个差异更明显行为数据库游标Kettle结果集读取方向可前可后仅向前行生命周期游标移动但不消费取走即消费跨转换传递需要额外连接由内存直接传递Kettle结果集做不到回退也没有FETCH FIRST这种语法。它唯一的控制参数是“每次执行获取行数”。这个参数决定一次从结果集里取多少行设成1一次取1行设成100一次取100行。在作业循环场景里这个值决定了目标转换每次执行拿到的是单行还是批量。想逐行处理就把值设为1并让作业按“每一行”执行想批量就调大值而不要勾选“执行每一行”。这个参数在Kettle使用教程里往往被一笔带过实际却直接影响跑批性能。2.3 结果集的传递范围同一个转换内还是跨转换结果集不是数据库表它跨步骤使用的边界是“同一次执行环境”。在同一个转换内部“复制行到结果”和“从结果获取记录”如果被设计成上下游关系理论上也能工作但Kettle的步骤调度是并行的。复制还没写完读取可能已经开始最终读到的行数不确定。所以我不建议在单个转换里做“复制到结果→从结果获取”这种自循环除非你在中间加“阻塞步骤”强制等待。更稳妥的做法是跨转换作业里先运行查询转换把结果集留在内存然后运行消费转换消费转换的第一个步骤就是“从结果获取记录”。这是标准姿势。你可以做一个最小实验验证作业里放两个转换A转换用“生成行”生成3行数据复制到结果B转换用“从结果获取记录”加“写日志”。不勾“执行每一行”运行作业B转换应该一次把3行全部打印出来。如果只打印一行或0行说明结果集没有传过去优先检查作业项连接是不是“成功”连接以及A转换是否真的执行到了“复制行到结果”。这一步能过滤掉一大半环境问题。结果集传递的范围搞清楚了循环才有的放矢。3. 循环获取结果集数据作业级循环和转换内循环怎么选循环的“循环体”到底是谁很多人没分清。在Kettle里有两种循环形态一种是作业级循环靠作业项之间的连接触发一种是转换内循环靠“从结果获取记录”配合脚本或插件实现。我强烈建议优先用作业级循环因为它更符合ETL流程的直观模型也更容易排错。转换内循环不是不能做但边界条件很多适合处理那些“不需要步骤间并行”的简单场景。3.1 作业级循环用作业连接属性实现“执行每一行”这是最主流、也最可控的做法。你需要两个转换查询转换负责查数据最后用“复制行到结果”把结果写进内存。消费转换第一个步骤必须是“从结果获取记录”后面接你要做的业务动作比如“表输入”“更新”“写日志”。然后在作业里把两个转换连起来右键连接线在弹窗里勾选“执行每一个输入行”。这一步就是循环的开关。完整的操作顺序是这样的新建作业添加“转换”作业项A选择查询转换。再添加“转换”作业项B选择消费转换。从A拉线到B编辑作业项连接勾选“执行每一个输入行”。在B转换的“从结果获取记录”里把“每次执行获取行数”设为1。运行作业观察日志。这里关键点是启用“执行每一个输入行”后作业调度器会把结果集的每一行作为一次独立执行上下文反复启动B转换直到结果集为空。所以A到B的连接本质上是一个循环表达式而不是普通的成功跳转。这种方式下B转换每次启动只处理一行非常干净。我在做“先查后处理”的批处理时基本都走这个模板。它的好处是查询只跑一次循环里不会重复查库消费转换可以独立调试如果某一行数据处理失败日志里能看到是在第几轮循环失败重跑成本低。3.2 转换内循环用“从结果获取记录”配合JavaScript的可行与不可行有人希望在单个转换里完成循环比如“表输入→复制行到结果→从结果获取记录→JavaScript代码”。这个方案理论上是可行的但有两个坑一是步骤并行执行时“复制行到结果”还没写完“从结果获取记录”就开始读了导致数据行数不稳定二是结果集被消费后不能回卷如果JavaScript代码里把行取完后面再没有数据可用。所以我很少在同一个转换里做这种自循环。如果只是简单处理可以在“JavaScript代码”步骤里直接读取当前结果集。但注意这个脚本并不是遍历一个独立结果集而是在作业循环上下文里每次从结果集取当前可读的行。示例// 在“JavaScript代码”步骤中读取当前结果集 var row; var resultRows []; // 从结果集取当前可读的行作业循环场景下通常是1行 row getRow(); while (row ! null) { // 取出字段默认按字符串处理 var orderId row.getString(order_id, ); processOrder(orderId); // 这里可以写你的业务逻辑 // 继续取下一行 row getRow(); } function processOrder(orderId) { // 打印到Kettle日志 log.logBasic(处理订单号: orderId); }这段代码的问题是getRow()并不保证你能拿到“结果集剩余的全部行”它在你单独运行转换时可能会立刻返回null因为当前流里没有输入行。所以在作业级循环里我更推荐直接用“从结果获取记录”把行变成流后面接“字段映射”或“表输出”不要硬用JavaScript。转换内循环只适合数据量小、且你已经验证过步骤调度顺序的场景。否则翻车成本比作业级循环高得多。3.3 把结果集数据传入转换的三种方式变量、参数、表输入结果集行进入目标转换后需要把它变成后续步骤能用的东西。常见有三种传法第一种是“设置变量”。在消费转换里加一个“设置变量”步骤把输入流里的字段写成Kettle变量作用范围选“在作业中使用”。后面同一个转换里的“表输入”“执行SQL脚本”就能用${order_id}这种占位符引用。第二种是“命名参数”。你在目标转换属性里定义参数然后在作业项“转换”的参数页签里映射。但这里有个容易误解的地方作业项的参数映射是静态的不能直接做到“结果集每行动态取值”。变量才能做到动态命名参数更适合在作业启动时传固定值。所以循环场景里我基本只用变量不硬套参数。第三种是“表输入”的SQL占位符。目标转换里的“表输入”步骤可以直接读取变量前提是勾选了“替换SQL里的变量”。例如SELECT * FROM trade_order WHERE order_id ${order_id}如果order_id是数字这样写没问题如果是字符串最好加引号SELECT * FROM trade_order WHERE order_id ${order_id}不加引号是导致SQL报“无效数字”或“列不存在”的常见原因。Kettle菜鸟教程里经常强调这一点但实际项目里还是会频繁看到。直接把“从结果获取记录”的输出流接到“更新”或“表输出”也可以这种方式更直观不用管变量作用域。缺点是你没法在SQL里做复杂拼接也没法把结果集字段再传给子转换。所以我一般用变量方式灵活性最高。4. 从结果集到转换传参变量映射、SQL占位和批量优化“把结果集数据传进去”这句话拆开看就是两个动作先把结果集字段写成变量再让目标SQL读取变量。这两个动作都有不少边界条件尤其是变量作用域和SQL占位符的替换时机。这一章把细节补齐。4.1 设置变量步骤的字段映射和类型细节结果集字段名不一定是目标SQL里需要的名字。比如查询转换输出的是ORDER_ID但目标转换里习惯叫order_id怎么办我不建议在“复制行到结果”之前的SQL里改别名因为那个查询转换可能被多个作业复用。更好的做法是在消费转换里做一次改名字。完整链路是从结果获取记录 → 字段选择 → 设置变量“字段选择”步骤可以做两件事改名和改类型。先把ORDER_ID改成order_id再把类型从默认的String或BigNumber显式转为Integer。这一步很重要因为Kettle变量本质上都是字符串后续SQL里如果直接比较数字数据库隐式转换通常能成功但遇到DECIMAL字段偶尔会变成科学计数法传过去后SQL结果完全不对。配置“设置变量”时的关键项字段名结果集里的字段名必须和“从结果获取记录”输出的字段名一致。变量名后续${}里写的名字。变量类型按需选String、Integer或Date。有效范围我通常选“在作业中使用”而不是“在JVM环境中”。选“在作业中使用”变量会跟着当前作业运行期走选“在JVM环境中”变量会留在JVM里可能导致第二次循环读到的还是上一次的值。这是一个很隐蔽的坑后面第5章会专门展开。4.2 在目标转换中替换变量${}与SQL的边界目标转换里能用变量的步骤包括“表输入”“执行SQL脚本”“更新”“删除”但它们都需要开启变量替换开关。以“表输入”为例步骤底部有一个“替换SQL语句中的变量”复选框如果没勾你会在日志里看到SQL原样输出SELECT * FROM trade_order WHERE order_id ${order_id}然后数据库报错找不到列或无效数字。我所有表输入步骤都会统一勾上“替换变量”。如果你用的是“执行SQL脚本”步骤也要在SQL里用${}并且确认勾选了“执行每一行”还是“替换变量”执行SQL脚本本身没有替换开关实际上它支持变量替换只要在作业或转换运行环境里变量存在。这里还有一个边界变量值如果含有单引号会导致SQL语法错误。比如cust_name变量值可能带ONeil直接替换后变成WHERE cust_name ONeil这种问题不能靠加引号解决需要在“设置变量”之前用“值映射”或“字符串替换”把单引号换掉。我在实际抽数里见过太多的翻车场景都是源数据里带了特殊字符。Kettle可以解析JSON吗可以但JSON里的引号嵌套又是另一个话题。总之变量替换适合干净的业务主键不适合大文本字段。4.3 批量循环调整“每次执行获取行数”而不是逐行如果结果集有5000行逐行执行目标转换会启动5000次转换光是转换启动开销就能让你等到怀疑人生。一个常见优化是调整“从结果获取记录”的“每次执行获取行数”把1改成500或1000并关掉作业连接上的“执行每一个输入行”。这样目标转换只执行一次或几次每次拿到一批行后续可以直接用“表输出”批量写库。我常用的参数组合场景每次执行获取行数是否勾选“执行每一行”适用场景逐行调用接口1是订单逐个更新状态批量写库500否结果集一次性或分批入库逐行跑复杂SQL1是每个ID单独执行存储过程批量然后组内循环100是每100行作为一个分组批次需要说明的是“执行每一行”的含义是“作业项会重复执行直到结果集为空”而“每次执行获取行数”决定每次执行能取到多少行。如果你勾选了“执行每一行”但又把每次获取行数设为500那么一次执行会拿到500行作业会循环执行直到结果集被取完。这种方式适合“一批一处理”的场景比如每500行更新一个批次号。如果你要调外部API建议还是保持逐行设成1并勾选“执行每一行”因为接口通常一次只接受一个参数。批量模式下调的API很容易被限流反而不如逐行稳。5. 避坑Kettle循环结果集最常见的6个翻车点这个功能看起来短实际跑起来坑不少。下面这6个问题是我在自建Kettle助手和日常跑批中反复踩出来的。5.1 现象“从结果获取记录”只取到一行现象目标转换确实执行了但“写日志”只打印最后一行或者只处理一行就停了。原因最常见的是作业项连接没有勾“执行每一个输入行”而“从结果获取记录”的“每次执行获取行数”又设成了1。这样目标转换只执行一次只从结果集里取一行其余行直接留在内存里作业结束就丢掉了。解决在作业连接属性里勾选“执行每一个输入行”或者把“每次执行获取行数”调大并让目标转换内部处理完所有行。如果你想逐行处理“每次执行获取行数”保持1但必须勾选“执行每一个输入行”。5.2 现象变量在目标转换里带不过去现象目标SQL里的${order_id}显示为空或者还是上一轮循环的旧值。原因设置变量的作用域选错了。选了“在JVM环境中”变量会在JVM里残留第二次循环启动时结果集里的新字段没有覆盖旧变量选了“在转换中”则在目标转换结束后变量就消失了后续步骤来不及使用。解决“设置变量”步骤的“有效范围”选“在作业中使用”并且确保每次循环执行目标转换时都会重新从结果集当前行取字段写变量。我一般把“设置变量”放在目标转换的最前面紧接着“从结果获取记录”这样能保证变量更新是串行的。5.3 现象作业循环第一次正常第二次结果集为空现象第一次运行作业循环处理了全部行第二次再跑发现消费转换一次都没执行日志里没有新数据。原因结果集被第一次循环消费掉了。Kettle结果集不是快照它是队列取走就没有了。查询转换虽然在作业里存在但如果只执行一次那么第二次跑作业时结果集并不会自动重新查询。解决把查询转换和消费转换放到同一个作业里并且让查询转换每次运行作业时都重新执行。如果你的业务确实需要“只查一次、反复循环”建议把查询结果落到临时表循环体从临时表读而不是从内存结果集读。临时表虽然慢一点但可控不会因为结果集被消费而消失。5.4 现象数据类型强制转换错误现象日志报错Couldnt convert value [abc] to Integer或提示Invalid byte sequence。原因结果集字段类型与实际内容不匹配。常见于数据库返回DECIMALKettle读成BigNumber在“设置变量”里转成String时变成科学计数法或者源字段是VARCHAR但里面是字母目标SQL按数字处理。解决在“从结果获取记录”后接“字段选择”显式设置每个要传参的字段类型。数字就用Integer或BigNumber日期就用Date并格式化。如果数据库是Oracle日期字符串要在SQL里写TO_DATE(${dateStr}, yyyy-mm-dd hh24:mi:ss)不要依赖数据库隐式转换。5.5 现象循环体内SQL执行特别慢现象结果集只有几十行但整个作业跑了很久每行都像重新登录数据库。原因目标转换每次循环都打开新的数据库连接没有用连接池导致连接开销占比远大于SQL执行本身。另一个原因是每次更新都自动提交事务开销大。解决在作业里配置“数据库连接”作业项勾选“使用连接池”让循环内转换共享同一个连接如果用的是“表输出”把“提交记录数量”调大比如1000条提交一次。这样循环性能能提高好几倍尤其适合几千行以上的循环。5.6 现象结果集字段名有中文或空格导致引用不到现象设置变量步骤的字段名下拉列表里找不到某个字段或者写日志打印出来是乱码。原因查询SQL里用了AS 编号这类中文别名Kettle结果集在后续步骤里对中文列名的支持不稳定尤其在跨字符集环境里经常翻车。解决所有结果集字段在查询SQL里都用英文别名例如SELECT order_id AS order_id FROM t_order。如果必须中文则在“从结果获取记录”之后立刻用“字段选择”改成英文字段名。不要在“设置变量”里硬填中文那不是解决之道。6. 验证循环结果集是否真的传入日志、行数与临时表三板斧我对这个功能最深的感受是不验证就等于没做。Kettle循环跑起来是黑色的你看不到内存里到底传了什么所以让关键信息落到日志和临时表里是非常值得养成的习惯。6.1 循环体前后打点用“写日志”确认每一行进入在消费转换的“从结果获取记录”后面接一个“写日志”步骤打印当前结果集行的主键。再在目标转换末尾也就是所有业务处理完成后的步骤后面加一个“写日志”打印“当前ID处理完成”。这样日志里会成对出现进入和完成记录。我通常在“写日志”里配置“打印字段”为order_id和系统当前时间能清楚看到循环次数和每行耗时。6.2 用“生成行”和临时表做冒烟测试不要一开始就连接生产库先用“生成行”生成10行测试数据生成行 → 复制行到结果 → 作业循环 → 从结果获取记录 → 写日志如果10行数据能在日志里完整打出10次“处理完成”说明循环配置正确。之后再把“生成行”换成生产环境的“表输入”排查范围就缩小到查询SQL本身。我还习惯在目标SQL里临时查一段变量替换后的结果-- 在目标转换里把变量用日志形式打印出来 SELECT ${order_id} AS debug_id FROM dual;然后接一个“写日志”步骤看输出是否真的是当前循环的ID。这个技巧能最快定位变量替换失效的问题。6.3 验证完成后别忘了清空结果集我最早做这个功能时总想把循环塞进一个转换里结果浪费时间。后来固定了“查询转换→复制行到结果→作业逐行→目标转换”这个模板并结合日志验证成功率非常高。最后补充一个习惯在作业末尾加一个“执行SQL脚本”步骤清理本次循环用到的临时表或者用空表重置结果集环境。因为Kettle结果集在作业结束后不一定立即释放同一个作业反复跑可能会被上一次的残留影响。清空临时结果集才算一个完整的循环闭环。希望这套验证方法和避坑经验能帮到你少走弯路。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
