第12章 Kettle实战

第12章：Kettle实战：文件到数据库ETL、多表关联、数据校验、作业调度与定时执行

说实话，很多初学者学Kettle，最怕的就是「学了一堆组件，真到项目里不知道怎么串起来」。我当年带团队的时候，有个新人把转换画得跟蜘蛛网一样，跑起来直接内存溢出。嗯，今天这一章，咱们就实打实地走一遍完整流程。

我会带你从文件读取开始，经过多表关联、数据校验，最后用作业调度让它自动跑起来。说白了，这就是一个生产级别的ETL小闭环。

12.1 文件到数据库：第一步怎么走？

先别急着拖组件。我习惯先搞清楚源文件长什么样。是CSV？Excel？还是固定宽度的文本？每种格式的坑都不一样。

举个例子，CSV文件最常见的坑就是字段里包含逗号。你想想看，如果某个字段值是"New York, NY"，Kettle默认按逗号切分，直接给你切成两列。我曾经因为这个原因，导致一整个批次的地址数据全部错位，排查了整整两个小时。

💡 我的习惯做法： 在「CSV文件输入」组件里，把「分隔符」改成制表符（\t），或者用「文本文件输入」组件，手动指定字段边界。如果必须用逗号，记得勾选「字符串限定符」为双引号。

配置步骤其实很简单：

拖入「CSV文件输入」或「Excel输入」组件
点击「浏览」选择文件，Kettle会自动预览前几行
检查字段类型——Kettle经常把数字识别成字符串，手动改一下
拖入「表输出」组件，配置数据库连接和目标表
用「字段选择」组件做映射，把源字段和目标字段对上

-- 目标表结构示例（MySQL）
CREATE TABLE ods_user_import (
    user_id      INT PRIMARY KEY,
    user_name    VARCHAR(50),
    email        VARCHAR(100),
    register_date DATE,
    source_file  VARCHAR(200),  -- 记录来源文件名
    create_time  DATETIME DEFAULT CURRENT_TIMESTAMP
);

这里有个小细节：我总会在目标表里加一个source_file字段。为什么？因为数据出问题的时候，你能快速定位是哪个文件导致的。嗯，这是血的教训换来的经验。

12.2 多表关联：别再用「数据库连接」了

很多人做多表关联，第一反应就是拖「数据库连接」组件。说实话，这个组件性能很差。我建议你优先用「流查询」或「合并连接」。

为什么？因为「数据库连接」每来一行数据，就去数据库查一次。如果源数据有10万行，它就查10万次。你想想看，这得多慢？

  实战场景： 假设我们要把用户订单表和商品信息表关联起来，生成一份订单明细报表。

我的做法是这样的：

主数据流：订单表（数据量大的那个）
查找流：商品信息表（数据量小的那个）
用「流查询」组件，把商品ID作为关联键
设置缓存大小（默认1000行，如果商品表有1万行，改成10000）

流查询配置要点：
- 查找字段：product_id
- 返回值：product_name, category, price
- 缓存大小：根据查找表行数设置，建议 = 查找表行数 × 1.2
- 是否区分大小写：看数据库排序规则

如果两个表都在同一个数据库里，而且数据量都很大（百万级以上），这时候用「数据库连接」反而更合适。为什么？因为数据库引擎自己会做优化，比Kettle逐行查询快得多。说白了，没有银弹，得看场景。

12.3 数据校验：别让脏数据进仓库

数据校验这块，我吃过不少亏。有一次，上游系统突然改了日期格式，从"2024-01-15"变成了"2024/01/15"，结果我的ETL直接报错，整批数据都没进去。嗯，从那以后，我养成了一个习惯：在入口处做校验。

Kettle里做校验，我推荐用这几个组件：

「字段校验」：内置了非空、唯一性、数据类型、长度等规则
「JavaScript代码」：复杂逻辑自己写，比如身份证号校验
「过滤记录」：把不符合条件的行分流到错误表

⚠️ 注意： 校验规则不要写死。我习惯把规则配置在数据库表里，ETL启动时动态读取。这样业务变了，改数据库就行，不用改ETL作业。

一个典型的校验流程：

用「字段校验」检查必填字段是否为空
用「过滤记录」把空值行分流到error_records表
用「唯一行」检查是否有重复数据
重复数据也写入错误表，并记录重复次数
干净的数据继续往下走，进入目标表

-- 错误记录表结构
CREATE TABLE etl_error_log (
    error_id     INT AUTO_INCREMENT PRIMARY KEY,
    job_name     VARCHAR(100),
    error_time   DATETIME,
    source_row   TEXT,          -- 原始数据行（JSON格式）
    error_type   VARCHAR(50),   -- 如：NULL_VALUE, DUPLICATE, FORMAT_ERROR
    error_desc   VARCHAR(500),
    status       CHAR(1) DEFAULT 'N'  -- N=未处理, Y=已处理
);

12.4 作业调度与定时执行：让它自己跑起来

手动跑ETL？那是新手干的事。我见过最夸张的，有人每天凌晨3点起来手动点一下「运行」。嗯，这其实可以用Kettle的作业（Job）来解决。

作业和转换的区别，说白了：

转换（Transformation）：做具体的数据处理，是「干活」的
作业（Job）：编排转换的执行顺序，是「管人」的

一个典型的作业流程：

「START」组件：作业入口
「检查文件是否存在」：判断源文件是否到位
「转换」组件：执行文件到数据库的ETL
「校验」组件：执行数据质量检查
「发送邮件」：成功或失败都发通知
「成功/失败」组件：结束

💡 定时执行方案： 我推荐用操作系统自带的调度工具，而不是Kettle内置的调度器。
- Linux：crontab
- Windows：任务计划程序
命令示例：0 2 * * * /opt/data-integration/kitchen.sh -file=/etl/jobs/daily_import.kjb

为什么不用Kettle自带的调度？因为一旦Kettle进程挂了，调度就停了。用crontab或任务计划程序，即使Kettle崩溃，操作系统也会帮你重启。我曾经因为Kettle内存溢出导致调度中断，整整三天没跑数据，后来就再也不用内置调度了。

12.5 避坑指南：我踩过的那些坑

最后，分享几个我亲身经历过的坑，希望能帮你少走弯路：

坑	现象	解决方案
文件编码问题	中文乱码	统一用UTF-8，在「文本文件输入」里指定编码
大文件内存溢出	Kettle卡死或OOM	开启「行集大小」限制，比如每次处理1000行
数据库连接池耗尽	连接超时	在作业里加「等待」组件，每次转换之间间隔2秒
字段类型不匹配	插入失败	用「字段选择」组件做显式类型转换

嗯，这一章的内容就到这里。记住一句话：ETL不是写代码，是搭积木。每个组件就是一块积木，关键是你得知道什么时候用哪一块，以及怎么把它们稳稳地搭在一起。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录