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執行即可。