ETL kettle使用详解

一：简介

ETL，是英文Extract-Transform-Load的缩写，用来描述将数据从来源端经过萃取（extract）、转置（transform）、加载（load）至目的端的过程。ETL一词较常用在数据仓库(Data Warehouse简称DW)，但其对象并不限于数据仓库。

ETL是将业务系统的数据经过抽取（Extract）、清洗转换（Cleaning、Transform）之后加载（Load）到数据仓库的过程，目的是将企业中的分散、零乱、标准不统一的数据整合到一起，为企业的决策提供分析依据。 ETL是商业智能（BI，Business Intelligence）项目重要的一个环节。

BI（Business Intelligence）即商务智能，它是一套完整的解决方案，用来将企业中现有的数据进行有效的整合，快速准确地提供报表并提出决策依据，帮助企业做出明智的业务经营决策。

二：Data Integration安装

2.1 下载 Data Integration

下载地址：https://sourceforge.net/projects/pentaho/files/Data Integration/ 选择最新的版本下载，这里最新版本为7.1，下载并解压 pdi-ce-7.1.0.0-12.zip 。注意：Data Integration是用Java开发的，是不区分系统的，Windows、MacOSX都是下载这个安装文件，这个安装文件是绿色版的，只需解压即可。

2.2 将mysql驱动放入到lib目录中

将mysql驱动jar(我这里使用的是5.1.40版本，mysql-connector-java-5.1.40.jar)放到pdi-ce-7.1.0.0-12/data-integration/lib目录下

三: 数据库准备

order库用于存储订单数据，user库用户存储用户信息数据，report库用于存储从order和user库关联的报表数据。

开发中我们将数据库分为两种类型：

一种是业务库，专门供业务调用；像表报业务等是不允许查询该数据库的。

另一种是报表库，我们要求报表中的数据不能查询业务库，我们需要将业务库中的数据保存到报表库中，然后从报表库中查询。

<code>order数据库
 
DROP
 
TABLE
 
IF
 
EXISTS
 
`tbl_goods`
;
CREATE
 
TABLE
 
`tbl_goods`
 (
 
`id`
 
bigint
(
20
) 
NOT
 
NULL
 AUTO_INCREMENT,
 
`name`
 
varchar
(
255
) 
DEFAULT
 
NULL
,
 
`price`
 
bigint
(
20
) 
DEFAULT
 
NULL
,
 PRIMARY 
KEY 
 (
`id`
)
) 
ENGINE
=
InnoDB
 AUTO_INCREMENT=
5
 
DEFAULT
 
CHARSET
=utf8;
INSERT
 
INTO
 
`tbl_goods`
 
VALUES
 (
'1'
, 
'iPhone X'
, 
'10'
), (
'2'
, 
'Apple Watch'
, 
'15'
), (
'3'
, 
'Macbook Pro'
, 
'20'
), (
'4'
, 
'AirPods'
, 
'30'
);
 
DROP
 
TABLE
 
IF 
 
EXISTS
 
`tbl_order`
;
CREATE
 
TABLE
 
`tbl_order`
 (
 
`id`
 
bigint
(
20
) 
NOT
 
NULL
 AUTO_INCREMENT,
 
`order_code`
 
varchar
(
64
) 
DEFAULT
 
NULL
,
 
`total_amount`
 
bigint
(
20
) 
DEFAULT
 
NULL
,
 
`create_time`
 datetime 
DEFAULT
 
CURRENT_TIMESTAMP
,
 
`user_id`
 
bigint 
(
20
) 
DEFAULT
 
NULL
,
 
`address_id`
 
bigint
(
20
) 
DEFAULT
 
NULL
,
 PRIMARY 
KEY
 (
`id`
)
) 
ENGINE
=
InnoDB
 AUTO_INCREMENT=
4
 
DEFAULT
 
CHARSET
=utf8;
INSERT
 
INTO
 
`tbl_order`
 
VALUES
 (
'1'
, 
'2019040314160001'
, 
'70'
, 
'2019-04-02 14:16:31'
, 
'1'
, 
'1' 
), (
'2'
, 
'2019040314160002'
, 
'130'
, 
'2019-04-03 17:34:58'
, 
'1'
, 
'1'
), (
'3'
, 
'2019040314160003'
, 
'340'
, 
'2019-04-03 18:22:07'
, 
'1'
, 
'1'
);
 
DROP
 
TABLE
 
IF
 
EXISTS
 
`tbl_order_item`
;
CREATE
 
TABLE
 
`tbl_order_item`
 (
 
`id`
 
bigint
(
20
) 
NOT
 
NULL 
 AUTO_INCREMENT,
 
`order_id`
 
bigint
(
20
) 
DEFAULT
 
NULL
,
 
`goods_id`
 
bigint
(
20
) 
DEFAULT
 
NULL
,
 
`quantity`
 
bigint
(
20
) 
DEFAULT
 
NULL
,
 
`price`
 
bigint
(
20
) 
DEFAULT
 
NULL
,
 PRIMARY 
KEY
 (
`id`
)
) 
ENGINE
=
InnoDB
 AUTO_INCREMENT=
7 
 
DEFAULT
 
CHARSET
=utf8;
INSERT
 
INTO
 
`tbl_order_item`
 
VALUES
 (
'1'
, 
'1'
, 
'1'
, 
'2'
, 
'10'
), (
'2'
, 
'1'
, 
'2'
, 
'4'
, 
'15'
), (
'3'
, 
'2'
, 
'1'
, 
'1'
, 
'10'
), (
'4'
, 
'2'
, 
'3'
, 
'6' 
, 
'20'
), (
'5'
, 
'3'
, 
'3'
, 
'2'
, 
'20'
), (
'6'
, 
'3'
, 
'4'
, 
'10'
, 
'30'
);/<code>

user数据库

<code> 
DROP
 
TABLE
 
IF
 
EXISTS
 
`tbl_address`
;
CREATE
 
TABLE
 
`tbl_address`
 (
 
`id`
 
bigint
(
20
) 
NOT 
 
NULL
 AUTO_INCREMENT,
 
`province`
 
varchar
(
255
) 
DEFAULT
 
NULL
,
 
`city`
 
varchar
(
255
) 
DEFAULT
 
NULL
,
 
`area`
 
varchar
(
255
) 
DEFAULT
 
NULL
,
 
`address`
 
varchar
(
255
) 
DEFAULT
 
NULL
,
 
`user_id`
 
bigint
(
20
) 
DEFAULT 
 
NULL
,
 PRIMARY 
KEY
 (
`id`
)
) 
ENGINE
=
InnoDB
 AUTO_INCREMENT=
2
 
DEFAULT
 
CHARSET
=utf8;
INSERT
 
INTO
 
`tbl_address`
 
VALUES
 (
'1'
, 
'上海'
, 
'上海市'
, 
'浦东新区'
, 
'张江镇xxx'
, 
'1'
);
 
DROP
 
TABLE
 
IF
 
EXISTS
 
`tbl_user`
;
CREATE
  
TABLE
 
`tbl_user`
 (
 
`id`
 
bigint
(
20
) 
NOT
 
NULL
 AUTO_INCREMENT,
 
`name`
 
varchar
(
255
) 
DEFAULT
 
NULL
,
 PRIMARY 
KEY
 (
`id`
)
) 
ENGINE
=
InnoDB
 AUTO_INCREMENT=
2
 
DEFAULT
 
CHARSET
=utf8;
INSERT
 
INTO
 
`tbl_user`
 
VALUES
 (
'1'
, 
'张三'
);/<code> 

report数据库

<code> 
DROP
 
TABLE
 
IF
 
EXISTS
 
`tbl_report_order_detail`
;
CREATE
 
TABLE
 
`tbl_report_order_detail`
 (
 
`id`
 
bigint
(
20
) 
NOT
 
NULL
 AUTO_INCREMENT,
 
`order_id`
 
bigint
(
20
) 
DEFAULT
 
NULL
,
 
`order_code`
 
varchar
(
255
) 
DEFAULT
 
NULL
,
  
`total_amount`
 
bigint
(
20
) 
DEFAULT
 
NULL
,
 
`goods_name`
 
varchar
(
255
) 
DEFAULT
 
NULL
,
 
`price`
 
decimal
(
10
,
0
) 
DEFAULT
 
NULL
,
 
`quantity`
 
varchar
(
255
) 
DEFAULT
 
NULL
,
 
`user_name`
 
varchar
(
255
) 
DEFAULT
  
NULL
,
 
`address`
 
varchar
(
255
) 
DEFAULT
 
NULL
,
 
`create_time`
 datetime 
DEFAULT
 
NULL
,
 PRIMARY 
KEY
 (
`id`
)
) 
ENGINE
=
InnoDB
 AUTO_INCREMENT=
39
 
DEFAULT
 
CHARSET
=utf8;/<code>

四：转换

0. etl在报表中的实践

etl一般在大数据、商务智能上使用较多，这里介绍一下使用etl来制作报表：从多个数据库中查询数据汇总在一起插入到一张单独的表中。

报表中的数据一般展示的比较全面，一般需要多个表关联查询才能拿到所有需要的数据。在开发中刚开始一般都是直接多表查询，随着微服务的出现，我们会对数据库进行拆分，一个微服务对应着一个数据库，比如订单微服务对应着订单库，用户微服务对应着用户库，这样会将表拆分到多个库中，报表中再进行多表关联查询就行不通了，为了解决拆库而不能多表连接查询的问题，我们使用Java代码来弥补这个缺陷，我们先在自己微服务对应的数据库先查部分数据，然后通过调用其它微服务的接口获取其它部分数据，然后再将这些部分数据组装成完整的数据，这样也能拿到完整的数据。有些报表查看是不使用任何条件过滤，即会走全表扫描的，随着订单数量的不断增多，随着部分查询不走索引，随着通过调用其它微服务接口来获取其它部分数据而消耗的时间，这些情况都会使得查询变得越来越慢，耗时越来越长，直到Mysql不能查询出来结果而超时。为了解决查询慢的问题，我们单独创建了一个报表，每个报表都在数据库中都对应着一张单独的表，当订单创建或者支付完成时会发送一个mq消息，mq系统接收到消息就将和该订单相关的数据查询出来然后插入到这个报表中，这样PC端调用接口查询报表数据只需要从这一张单表中去查询就可以了，单表查询是非常快的。发送mq的解决思路可以解决Mysql查询过慢的问题，但是需要在mq系统中首先要写java代码去查询相关数据，然后再插入单表中，如果报表数据改变了(增加了新的字段)还需要更新这部分逻辑，为了使当报表中有数据变更时不修改代码，我们使用了etl技术。

1. 双击Data Integration启动程序

2. 点击 主对象树 中的 转换

3. 将 输入：表输入 拖到右边

4. 双击转换中的表输入

• 输入步骤名称
• 数据库连接 新建…
• 输入SQL语句
• 选中 替换SQL语句中的变量

<code>
SELECT 

	o.id 
AS
 order_id,
	o.order_code,
	o.create_time,
	o.user_id,
	o.total_amount,
	tg.
`name`
 
AS
 goods_name,
	toi.price,
	toi.quantity
FROM
 tbl_order_item toi
LEFT
 
JOIN
 tbl_goods tg 
ON
 toi.goods_id = tg.id
LEFT
 
JOIN
 tbl_order o 
ON
 toi.order_id = o.id
WHERE
 o.create_time >= 
'${beginTime}'
 
AND
 o.create_time <= 
'${endTime}'
/<code>

5. 设置环境变量

菜单栏： 编辑-> 设置环境变量

6. 再拖入一个输入：表输入

<code>
SELECT
 
	tu.id 
AS
 uid,
	tu.
`name`
 
AS
 user_name,
	
concat
(ta.province, ta.city, ta.area, ta.address) 
AS
 address
FROM
 tbl_user tu
 
LEFT
 
JOIN
 tbl_address ta 
ON
 tu.id = ta.user_id/<code>

7. 拖入一个 记录关联(笛卡尔输出)

将鼠标悬浮到表输入，然后点击向右的箭头，然后在记录关联(笛卡尔输出)上单机一下，将表输入和记录关联连接起来。同理将表输入(user)也和记录关联(笛卡尔输出)连接起来。

这里是模拟Left Join, 如果要是模拟Union Join 则使用 合并记录。

8. 双击记录关联(笛卡尔输出)

选择表关联的主表，选择关联的字段

9. 拖入一个流程：过滤记录

将记录关联(笛卡尔输出)和过滤记录连接起来，双击过滤记录。

过滤记录不是必须的，这里只是演示还可以对结果进行过滤筛选。

10. 拖入一个输出：插入/更新

将过滤记录和 插入/更新连接在一起

数据库连接 新建, 数据库名称为 report

点击获取字段，这里的字段为联合主键字段，即能够唯一确定一条记录的字段，这里通过order_id, user_name, goods_name 能够确定一条唯一的记录，如果联合主键存在这样的记录则更新数据，否则插入数据。这里查询的字段也可以是所有字段

点击获取和更新字段，删除user_id、uid

11. 执行转换

五：作业

运行这个转换只能执行一次，kettle还支持定时调度作业的形式执行定时任务。

点击作业，将通用: START和通用: 转换 拖拽到右侧窗口，然后将START和转换连接起来。然后配置一些START的定时执行时间，配置一下转换的对应的.ktr文件，最后点击点击左上角的三角箭头Run执行即可。