简介这是一套基于KettlePentaho Data Integration二次开发的Web版数据集成平台源码包面向数据工程师、ETL开发者及希望降低数据采集门槛的分析人员。它把传统桌面端Kettle的转换与作业能力搬到浏览器中通过拖拽式可视化界面完成数据源接入、清洗转换与任务调度解决非技术人员难以使用ETL工具、团队数据集难以共享复用的问题。压缩包共1645个文件约160.93MB以912个Java源码、163个properties配置、123个xml、94个css与74个vue前端组件为主另含ktr转换文件、Dockerfile、shell脚本及yaml部署配置覆盖前后端与容器化部署全链路。目前已有727人学习下载。读者可从中了解Kettle如何被封装为Web服务掌握数据源管理、图形化转换设计、执行监控、版本控制与权限角色等模块的实现思路并可直接基于源码扩展定制适合作为数据集成平台的学习与二次开发参考。1. 从 Kettle 到 Web 拖拽为什么值得把 ETL 搬到浏览器里很多团队第一次接触 Kettle 都是在本地装 PDIPentaho Data Integration用 Spoon 图形界面拖几个步骤配好数据库连接点运行数据就从 A 库流到了 B 库。这套流程在单机时代非常顺手但一旦要多人协作、要定时调度、要把任务交给非技术同事维护问题就来了Spoon 是桌面客户端转换文件散落在各人电脑上版本对不上谁改了哪一步说不清楚。于是「基于 Kettle 实现的 Web 版数据集成平台」这个方向就有了真实需求——把 Kettle 的转换和作业能力放到浏览器里用可拖拽的画布编排数据流后端统一调度执行。这个标题拆开看有三个关键词Kettle、Web、可拖拽。Kettle 负责底层 ETL 引擎Web 负责访问方式和多人协作拖拽负责降低使用门槛。适合谁适合手里已经有一堆 Kettle 转换脚本、想把它平台化的数据工程师也适合想给业务方提供一个「自己拖几个算子就能跑数」入口的团队。下面按「引擎怎么接、画布怎么做、任务怎么跑、坑在哪」的顺序讲清楚。2. 把 Kettle 引擎嵌进 Web 服务从 Spoon 到 API 的改造路径2.1 为什么不能直接调 Spoon而要用 Kettle 的 Java APISpoon 是 Kettle 的桌面客户端它本身也是基于 Kettle 核心引擎kettle-core、kettle-engine构建的。Web 平台要做的不是启动 Spoon而是直接依赖 kettle-core 和 kettle-engine 这两个 jar在服务端用 Java 代码加载转换文件、设置参数、执行并收集日志。常见做法是引入 Maven 依赖然后通过TransMeta、Trans、StepMeta这些类来操作转换。选型理由很直接Spoon 的界面逻辑和 Web 画布完全是两套东西硬套只会把桌面端的复杂度带进来。而 Kettle 的引擎 API 是稳定的转换文件.ktr和作业文件.kjb本质是 XMLWeb 端只需要生成和解析这些 XML就能和引擎对接。这样前端画布拖出来的节点最终序列化成 Kettle 能识别的 XML后端调用引擎执行职责清晰。2.2 最小可运行示例用 Java 代码加载并执行一个转换下面这段代码演示了服务端如何加载一个已有的 .ktr 文件并执行。实际平台里这个 .ktr 可能是前端画布保存后生成的也可能是用户上传的。import org.pentaho.di.core.KettleEnvironment; import org.pentaho.di.core.exception.KettleException; import org.pentaho.di.trans.Trans; import org.pentaho.di.trans.TransMeta; public class KettleRunner { public static void main(String[] args) throws KettleException { // 初始化 Kettle 环境只需执行一次 KettleEnvironment.init(); // 加载转换文件这里假设文件在 resources 目录下 TransMeta transMeta new TransMeta(path/to/your_transformation.ktr); // 创建转换实例 Trans trans new Trans(transMeta); // 设置命名参数对应转换里定义的变量 trans.setParameterValue(SOURCE_DB, jdbc:mysql://localhost:3306/source); trans.setParameterValue(TARGET_DB, jdbc:mysql://localhost:3306/target); // 执行转换 trans.execute(null); // 等待执行完成 trans.waitUntilFinished(); // 检查执行结果 if (trans.getErrors() 0) { System.err.println(转换执行出错错误数 trans.getErrors()); } else { System.out.println(转换执行成功); } } }逻辑说明KettleEnvironment.init()会加载 Kettle 的插件和配置必须在任何 Kettle 操作之前调用且整个 JVM 生命周期内只调一次。TransMeta负责解析 .ktr 文件Trans是执行实例。setParameterValue用于传入运行时参数这样同一个转换文件可以复用到不同环境。execute(null)启动执行waitUntilFinished()阻塞直到结束。参数方面trans.getErrors()返回错误计数平台里应该把它和日志一起落库方便排查。2.3 转换 XML 的生成与解析前端画布和后端的契约前端拖拽画布最终要产出 Kettle 能识别的 XML。一个转换的 XML 结构大致是transformation根节点下面有info存元信息step定义每个步骤hop定义步骤之间的连线。前端每拖一个节点就对应一个step每连一条线就对应一个hop。常见做法是后端定义一套 DTOData Transfer Object前端画布用 JSON 描述节点和连线保存时后端把 JSON 转成 Kettle XML。这样做的好处是前端不用关心 XML 细节后端可以统一做校验和版本兼容。下面是一个简化的 JSON 到 XML 的映射示例{ name: demo_trans, steps: [ { name: 表输入, type: TableInput, config: { connection: source_db, sql: SELECT id, name FROM users } }, { name: 表输出, type: TableOutput, config: { connection: target_db, table: users_copy } } ], hops: [ { from: 表输入, to: 表输出 } ] }后端拿到这个 JSON 后用 DOM 或 JAXB 生成对应的step和hop节点。参数说明type必须和 Kettle 插件类型名一致比如TableInput、TableOutput、SelectValues等写错了引擎会报找不到步骤插件。connection对应 Kettle 里定义的数据库连接名平台需要单独维护连接管理模块。提示Kettle 的步骤类型名区分大小写建议在平台里维护一份允许的步骤类型白名单前端只暴露白名单内的节点避免用户拖出无法执行的步骤。3. Web 拖拽画布的实现前端选型与节点连线逻辑3.1 画布库选型为什么常见方案是 AntV X6 或 LogicFlowWeb 拖拽画布不是简单的 div 拖拽它需要处理节点渲染、连线锚点、缩放平移、框选、撤销重做等。常见做法是用成熟的图编辑库国内团队用得比较多的是 AntV X6 和 LogicFlow。X6 的文档和示例比较全支持自定义节点和连线适合做数据集成这种节点类型固定的场景LogicFlow 更轻量扩展性也不错。选型时要考虑几个点是否支持自定义 HTML 节点因为 Kettle 步骤需要展示图标、名称、状态、是否支持连线校验比如不允许环形连接、是否支持导出 JSON。X6 在这些方面都比较成熟下面以 X6 为例讲节点拖拽和连线。3.2 从左侧面板拖拽节点到画布最小实现左侧面板列出可用的 Kettle 步骤类型用户按住拖到画布上画布在落点创建一个节点。X6 提供了Dnd插件来简化这个流程。import { Graph, Dnd } from antv/x6; // 初始化画布 const graph new Graph({ container: document.getElementById(canvas), grid: true, connecting: { router: manhattan, connector: rounded, allowBlank: false, allowLoop: false, allowMulti: false, }, }); // 初始化拖拽插件 const dnd new Dnd({ target: graph, scaled: false, animation: true, }); // 左侧面板的节点模板 const nodeList [ { type: TableInput, label: 表输入, icon: table }, { type: TableOutput, label: 表输出, icon: output }, { type: SelectValues, label: 字段选择, icon: select }, ]; // 渲染左侧面板并绑定拖拽事件 nodeList.forEach((item) { const div document.createElement(div); div.className node-item; div.textContent item.label; div.addEventListener(mousedown, (e) { const node graph.createNode({ shape: rect, width: 120, height: 40, label: item.label, data: { type: item.type }, }); dnd.start(node, e); }); document.getElementById(palette).appendChild(div); });逻辑说明Dnd插件负责把左侧的节点模板拖到画布上graph.createNode创建节点时把 Kettle 步骤类型存在data.type里后续保存时根据这个类型生成 XML。connecting配置里allowLoop: false禁止自环allowMulti: false禁止重复连线router: manhattan让连线走直角视觉上更像数据流。参数说明width和height是节点尺寸实际平台里可以根据步骤类型显示不同图标和端口。allowBlank: false表示连线不能悬空必须连到节点上。这些配置直接影响用户体验建议在画布初始化时就定好后期改起来成本高。3.3 连线校验与数据流方向避免用户拖出跑不通的转换Kettle 的转换是有向无环图步骤之间通过 hop 连接数据从上游流向下游。如果用户连出一个环或者把输出步骤连到输入步骤引擎执行时会报错。所以画布层要做校验连线时检查是否形成环、是否违反步骤的输入输出约束。常见做法是给每个步骤类型定义允许的输入端口数和输出端口数。比如「表输入」只有输出端口「表输出」只有输入端口「字段选择」既有输入也有输出。连线时判断源节点的输出端口和目标节点的输入端口是否匹配同时用图算法检测是否成环。// 连线校验示例 graph.on(edge:connecting, ({ edge }) { const source edge.getSourceCell(); const target edge.getTargetCell(); if (!source || !target) return; // 检查是否成环 if (wouldCreateCycle(graph, source.id, target.id)) { edge.remove(); alert(不允许形成环形连接); return; } // 检查端口类型是否匹配 const sourceType source.getData().type; const targetType target.getData().type; if (!isPortCompatible(sourceType, targetType)) { edge.remove(); alert(步骤连接不兼容); } }); function wouldCreateCycle(graph, sourceId, targetId) { // 从 target 出发做深度优先搜索如果能回到 source 则成环 const visited new Set(); const stack [targetId]; while (stack.length) { const current stack.pop(); if (current sourceId) return true; if (visited.has(current)) continue; visited.add(current); const outgoing graph.getOutgoingEdges(current) || []; outgoing.forEach((edge) stack.push(edge.getTargetCellId())); } return false; }逻辑说明edge:connecting事件在连线过程中触发此时可以拦截并校验。wouldCreateCycle用深度优先搜索判断新连线是否会导致环。isPortCompatible根据步骤类型判断端口是否匹配这个函数需要平台维护一份步骤元数据表。参数说明graph.getOutgoingEdges返回从某个节点出发的所有边遍历时注意去重。实际平台里校验失败应该给出更友好的提示而不是简单 alert比如高亮冲突的节点。注意前端校验只是第一道防线后端保存转换时还要再校验一次因为用户可能绕过前端直接调接口。后端可以用 Kettle 的TransMeta加载 XML检查是否有环和非法步骤。4. 任务调度与执行隔离Web 平台怎么跑 Kettle 转换4.1 调度模型定时、手动、依赖触发三种方式Web 平台上的 Kettle 转换需要被触发执行常见有三种方式手动点击运行、定时调度类似 crontab、依赖触发上游转换完成后触发下游。手动运行最简单前端点按钮后端调 Kettle API 执行。定时调度需要引入调度框架常见做法是用 Quartz 或 XXL-JOB把转换 ID 和 cron 表达式存库调度器到点触发执行。依赖触发适合有上下游关系的转换比如先抽数再清洗再入库。实现方式可以是在转换执行完成后查询依赖它的下游转换依次触发。这里要注意避免循环依赖平台里应该维护一张依赖关系图保存时做环检测。4.2 执行隔离每个转换独立 ClassLoader 还是共享Kettle 的插件机制依赖 ClassLoader如果多个转换在同一个 JVM 里并发执行可能会遇到插件冲突或资源竞争。常见做法是每个转换执行时创建一个独立的 ClassLoader或者至少把 Kettle 的初始化放在应用启动时完成执行时只创建Trans实例。独立 ClassLoader 的好处是隔离性好不同转换可以用不同版本的插件坏处是内存占用高创建 ClassLoader 本身也有开销。对于大多数平台共享 Kettle 环境、每个转换创建独立Trans实例已经够用。如果确实需要隔离可以考虑把执行器拆成独立进程Web 层只负责调度和状态查询。// 执行器伪代码提交任务到线程池 ExecutorService executor Executors.newFixedThreadPool(10); public void submitTrans(Long transId, MapString, String params) { executor.submit(() - { try { TransMeta transMeta transMetaService.load(transId); Trans trans new Trans(transMeta); params.forEach(trans::setParameterValue); trans.execute(null); trans.waitUntilFinished(); // 记录执行日志和状态 executionLogService.save(transId, trans.getErrors(), trans.getLogChannelId()); } catch (Exception e) { executionLogService.saveError(transId, e.getMessage()); } }); }逻辑说明用线程池控制并发数避免同时跑太多转换把数据库连接打满。trans.getLogChannelId()可以拿到日志通道 IDKettle 的日志可以配置输出到数据库或文件平台里通常会把日志收集起来展示给用户。参数说明线程池大小要根据数据库连接池和服务器配置调整一般建议不超过数据库最大连接数的三分之一。trans.setParameterValue传入的参数会覆盖转换里定义的默认值。4.3 日志收集与执行状态回传Kettle 执行过程中会产生大量日志平台需要把这些日志收集起来按执行实例存储并提供查询接口。常见做法是实现一个自定义的LogChannelInterface或者用 Kettle 的TransListener监听执行事件把日志写入数据库或消息队列。前端展示时可以轮询执行状态接口或者用 WebSocket 推送日志。对于长任务建议用 WebSocket避免频繁轮询。执行状态一般有等待中、运行中、成功、失败、已停止。停止功能需要调用trans.stopAll()注意这个方法会停止所有步骤可能留下未完成的数据。提示Kettle 的日志级别可以在转换里配置平台里建议默认用Basic级别需要排查时再调成Detailed。日志量太大会影响性能尤其是写入数据库时。5. 避坑与排查Web 化 Kettle 平台最常见的 5 个翻车点5.1 现象转换在 Spoon 里能跑Web 平台上报「找不到插件」原因Web 服务端的 Kettle 环境没有加载对应的插件或者插件版本和转换文件里记录的不一致。Kettle 的步骤类型依赖插件注册如果服务端缺少某个插件 jarTransMeta解析时会报错。解决确保服务端的kettle-engine和kettle-core版本与转换文件兼容并且所有用到的插件都在 classpath 里。常见做法是把 Kettle 的plugins目录整体打包进应用或者用 Maven 引入需要的插件依赖。平台里维护一份步骤类型白名单前端只暴露服务端支持的步骤。5.2 现象前端拖拽保存后转换执行报「hop 连接无效」原因前端生成的 XML 里hop的from和to属性引用的步骤名称和step里的name不一致或者步骤顺序不对。Kettle 对 hop 的引用是大小写敏感的名称必须完全匹配。解决保存时后端做一次校验用TransMeta加载生成的 XML检查 hop 引用的步骤是否存在。前端生成 XML 时步骤名称建议用唯一 ID 而不是用户可见的标签避免重名和特殊字符问题。5.3 现象并发执行多个转换时数据库连接池被打满原因每个 Kettle 转换执行时会创建自己的数据库连接如果并发数太高连接池不够用后续转换会等待或失败。Kettle 的数据库连接是在步骤初始化时创建的执行完成后不一定立即释放。解决限制并发执行的转换数量用线程池控制。同时配置 Kettle 的连接池参数比如trans.setUsingThreadPriorityManagment(false)减少线程竞争。数据库连接建议用连接池管理避免每个转换都新建物理连接。5.4 现象转换执行成功但数据没写进去日志里也没有报错原因Kettle 的「表输出」步骤默认是批量提交如果转换结束时没有正确提交事务数据可能留在缓冲区。另外如果步骤配置了「不执行」或者条件分支没走到也会出现这种情况。解决检查「表输出」步骤的提交记录数设置确保转换结束时事务提交。平台里可以在转换执行完成后主动调用trans.commit()或者检查步骤的getLinesOutput()计数。日志级别调到Detailed可以看到每个步骤的读写行数。5.5 现象Web 画布上节点多了之后拖拽卡顿原因X6 或 LogicFlow 在节点数量多时渲染和事件处理会变慢。如果每个节点都用复杂的 HTML 自定义渲染性能下降更明显。解决节点渲染尽量用 SVG 而不是 HTML减少 DOM 操作。画布开启虚拟渲染或分片渲染X6 支持virtual: true。另外连线校验的图算法不要在每次鼠标移动时都执行可以加防抖或者只在连线完成时校验。6. 进阶技巧用转换模板和参数化让平台真正可复用平台做到后面用户不会满足于每次从零拖一个转换。更常见的需求是把常用的转换存成模板下次直接拖模板出来改参数就能用。Kettle 本身支持命名参数在转换里用${PARAM_NAME}引用执行时通过trans.setParameterValue传入。平台可以把模板和参数定义一起存库用户实例化模板时只填参数值。另一个技巧是转换的版本管理。每次保存生成一个新版本执行时记录用了哪个版本出问题可以回滚。Kettle 的 .ktr 文件是文本适合做 diff平台可以展示版本之间的差异方便排查「谁改了哪一步」。-- 转换版本表设计示例 CREATE TABLE trans_version ( id BIGINT PRIMARY KEY AUTO_INCREMENT, trans_id BIGINT NOT NULL, version INT NOT NULL, content TEXT NOT NULL, -- 存储 Kettle XML params JSON, -- 参数定义 created_by VARCHAR(64), created_at DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_trans_version (trans_id, version) );逻辑说明每次保存转换时插入一条新版本记录content存完整的 Kettle XMLparams存参数定义。执行时根据trans_id和version加载对应的 XML。这样即使用户改了转换历史执行记录仍然能追溯到当时的版本。参数说明version从 1 开始递增可以用SELECT MAX(version) 1生成但要注意并发保存时的竞争建议用数据库唯一索引兜底。params用 JSON 类型存储方便前端渲染参数表单。我自己踩过最深的坑是早期没做版本管理用户改完转换直接覆盖结果某次调度失败后完全不知道之前跑的是哪个版本排查花了整整一个下午。从那以后任何平台化的 ETL 工具我都会先把版本表建好再动别的。希望帮到你。本文还有配套的精品资源点击获取