利用Kettle進行資料同步(下)

NO IMAGE

版權宣告:
本文為博主原創文章,未經博主允許不得轉載。關注公眾號技術匯(ID: jishuhui_2015) 可聯絡到作者。

上篇介紹了基於kettle的資料同步工程的搭建,entrypoint.kjb就是整個工程執行的入口。

為了進一步降低操作成本,讓整個資料同步過程更穩定、安全,需要進行更高層面的抽象,做成一個簡單易用的系統。

以下是應用截圖:

效果圖

除了選擇資料來源和資料庫之外,還加入了授權碼,意味著授權範圍內的使用者才能使用該系統。

因為是內部使用,授權使用者還沒實現後臺管理,直接往應用資料庫裡新增,所選擇的資料來源和資料庫都是通過配置檔案生成的。

文末會附上GitHub上的原始碼地址,有需要的讀者,可以進行二次開發改造。

一、資料庫設計

資料庫名稱:kettle,目前有兩張表:

1、授權使用者表。表內記錄的使用者即可使用資料同步系統

CREATE TABLE `authorized_user` (
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT '使用者ID,自增',
`user` varchar(128) NOT NULL COMMENT '使用者名稱,全域性唯一',
`token` varchar(20) NOT NULL COMMENT '使用者的授權碼,全域性唯一',
`status` char(1) NOT NULL DEFAULT 'A' COMMENT '授權使用者狀態:A-已授權,R-未授權',
`gmt_create` datetime NOT NULL COMMENT '建立時間',
`gmt_modify` datetime NOT NULL COMMENT '最後修改時間',
PRIMARY KEY (`id`),
UNIQUE KEY `unique_index_token` (`token`) USING BTREE,
UNIQUE KEY `unique_index_user` (`user`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='授權使用者表'

2、同步記錄表。記錄使用者的資料同步操作

CREATE TABLE `sync_record` (
`sync` varchar(20) NOT NULL COMMENT '同步記錄主鍵',
`ipv4` varchar(15) NOT NULL COMMENT 'ip地址',
`from_db` varchar(100) NOT NULL COMMENT '源資料',
`to_db` varchar(100) NOT NULL COMMENT '目標資料',
`user` varchar(128) NOT NULL COMMENT '使用者名稱',
`token` varchar(20) NOT NULL COMMENT '使用者的授權碼',
`status` char(1) NOT NULL DEFAULT 'P' COMMENT '同步狀態:P-正在執行,S-成功,F-失敗',
`gmt_create` datetime NOT NULL COMMENT '同步建立時間',
`gmt_modify` datetime NOT NULL COMMENT '最後修改時間',
PRIMARY KEY (`sync`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='同步記錄表';

二、程式設計

因為系統做得比較簡單實用,沒有什麼特別設計之處。筆者重點說三點:

1、資料來源及其引數配置

在application.yml配置檔案中,有這麼一段配置:

env:
entry-point: kettle/entrypoint.kjb
databases:
- taxi-user
- taxi-account
- taxi-trade
- taxi-coupon
- taxi-bi
- taxi-system
- taxi-credits
- taxi-finance
- taxi-notification
- taxi-gateway
from-dbs:
- PROD
- TEST
- LOCAL
to-dbs:
- LOCAL
- TEST
db-settings:
- name: LOCAL
host: *****
port: 3306
user: *****
password: *****
- name: TEST
host: *****
port: 3306
user: *****
password: *****
- name: PROD
host: *****
port: 3306
user: *****
password: *****

利用了springboot的@ConfigurationProperties的註解。

@Setter
@Getter
@ConfigurationProperties(prefix = "env")
public class EnvConfig {
private List<String> databases;
private List<String> fromDbs;
private List<String> toDbs;
private List<DBSetting> dbSettings;
public DBSetting getDBConfig(String name) {
if (StringUtils.isBlank(name)) return null;
return dbSettings.stream().filter(dbSetting -> dbSetting.getName().equalsIgnoreCase(name)).findFirst().orElse(null);
}
}

當中的DBSetting的定義如下所示:

@Setter
@Getter
@NoArgsConstructor
public class DBSetting {
private String name;
private String host;
private String port = "3306";
private String user = "root";
private String password;
public DBSetting(String host, String user, String password) {
this.host = host;
this.user = user;
this.password = password;
}
}

通過客戶端傳來的引數,即可定位到對應的引數設定。

2、整合kettle的API
因為kettle相關的jar包放在了自己搭建的nexus私服上,所以如果使用的是maven管理jar包的話,需要在settings.xml配置檔案中做一點修改:

<mirror>
<id>nexus</id>        
<url>公司內部的nexus的URL</url>       
<mirrorOf>*,!pentaho-releases</mirrorOf>       
</mirror> 

其中的mirrorOf節點加上了!pentaho-releases,表示排除pentaho-releases。

然後,在springboot工程中的pom.xml中指定pentaho-releases的url。

<repositories>
<repository>
<id>pentaho-releases</id>
<url>https://nexus.pentaho.org/content/groups/omni/</url>
</repository>
</repositories>

接下來是核心的對接程式碼,具體可以參照工程原始碼。

JobMeta jobMeta = getJobMeta(new ClassPathResource(envConfig.getEntryPoint()));
Job job = new Job(null, jobMeta);
//設定Variable
job.setVariable("sync", sync);
job.setVariable("TO_HOST", toDbSetting.getHost());
job.setVariable("TO_DB", form.getDb());
job.setVariable("TO_USER", toDbSetting.getUser());
job.setVariable("TO_PASSWORD", toDbSetting.getPassword());
job.setVariable("TO_PORT", toDbSetting.getPort());
job.setVariable("FROM_HOST", fromDbSetting.getHost());
job.setVariable("FROM_DB", form.getDb());
job.setVariable("FROM_USER", fromDbSetting.getUser());
job.setVariable("FROM_PASSWORD", fromDbSetting.getPassword());
job.setVariable("FROM_PORT", fromDbSetting.getPort());
job.start();              //開始執行Job
job.waitUntilFinished();  //等待Job完成

3、非同步執行作業

因為一個Job的執行時間可能會很長,這個主要是看資料量的多少,所以一個request的來回可能會導致TIMEOUT,所以需要改為非同步的模式。

其核心的思想是:啟動新的執行緒,客戶端定時輪詢執行結果。

三、總結

筆者分兩篇文章介紹瞭如何利用kettle進行資料同步,並實現一個簡易的系統,降低操作成本和出錯率。

就介紹到這了,如有疑問,可以留言。

歡迎fork我的工程程式碼(https://github.com/liu-weihao/kettle)。

關注我們