乾貨 | 快速實現數據導入及簡單DCS的實現

乾貨 | 快速實現數據導入及簡單DCS的實現

原創： 趙琦 京東雲開發者社區  4月18日

對於多數用戶而言，在利用雲計算的大數據服務時首先要面臨的一個問題就是如何將已有存量數據快捷的導入到大數據倉庫中。本文將演示如何基於京東雲數據計算服務平臺，簡單、快速地將數據導入數據計算服務。

咱們一般說的大數據平臺主要包括三部分：數據相關的產品和技術、數據資產、數據管理。京東雲數據計算服務（Data Computing Service，簡稱：DCS）是一個全託管、低使用成本的雲上數據倉庫服務。經過數據工廠，可輕鬆實現雲上各數據源(包括對象存儲、雲數據庫、數據倉庫等)間、本地數據源與雲數據源間的多種不一樣數據源的數據同步，實現多源數據分析與管理。

在數據工廠服務中能夠建立同步任務來搬運數據，並按照指定的調度策略（天天、每週、每個月）運行。該模塊提供任務監控和告警功能，用戶能夠經過任務執行的明細日誌和告警歷史，輕鬆查明問題所在；同時，提供全面託管的工做流服務，支持圖形化設計數據分析。以工做流任務的方式實現對數據的處理和相互依賴，幫助用戶快速構建數據處理分析做業並週期性地執行。

下面會以MySQL數據庫爲例說明如何利用京東雲數據工廠進行數據採集和DTS數據庫之間的數據同步。數據工廠支持常見RDS數據庫，如MySQL、SQLServer、Oracle、DB2和NoSQL數據庫，也支持從OSS、FTP站點及Elastic Search等。

下圖是數據工廠支持的數據源種類：

下面會演示以MySQL數據庫爲例，如何利用數據工廠進行數據採集，以及如何利用數據工廠做爲DTS在兩個數據源之間進行數據遷移。

爲了方便測試，首先咱們要建立數據庫表單並灌入測試數據。爲了測試方便，提早建立了一臺CentOS 7.4雲主機做爲模擬客戶端對數據庫進行訪問。

1、核心概念

數據集

在數據工廠服務中的數據集，是指由同步任務時需指定的數據源端或數據目標端的不一樣數據存儲實例。所以，在建立同步任務以前，你必須先鏈接數據集。同一個數據集，能夠是多個同步任務的數據源端或者數據目標端。

數據集的連通性

在用戶建立數據集鏈接時，爲檢驗數據集成服務可否鏈接成功，須要用戶更具根據不一樣的數據集類型，填入相應的值用於連通性校驗。數據集鏈接成功，是數據同步任務成功的前提。

同步任務

同步任務，是用戶使用數據集成服務的最小單元。每個同步任務須要用戶配置數據源端、數據目標端以及相應的同步策略（如髒數據的處理等）。

工做流

工做流以圖形化設計任務的方式實現對數據的處理和相互依賴。

開始實戰

1、準備測試數據源

首先建立模擬數據環境。本例採用京東雲的RDS MySQL 8.0數據庫服務做爲數據源，建立時可指定數據庫名字爲Testdb。數據庫建立完成後要開啓外網訪問，數據工廠能夠經過公網IP或域名對數據庫進行訪問，詳細域名能夠在數據庫的詳情頁中找到。

建立MySQL數據庫能夠經過圖形界面按提示填寫必要信息開通，這裏不贅述。須要提醒的是數據庫開通後默認不容許外網訪問，要點擊開啓外網訪問，並記住默認端口3306。

對MySQL訪問能夠經過圖形界面訪問或經過客戶端訪問。固然也能夠經過其餘支持MySQL的圖形化客戶端進行訪問。

本例使用CentOS 系統做爲客戶端訪問，如未安裝客戶端可使用Yum命令安裝 MySQL。順利安裝能夠看到以下提示。

 1[root@CentOS ~]# yum install mysql
 2Loaded plugins: fastestmirror, langpacks
 3Loading mirror speeds from cached hostfile
 4base                                | 3.6 kB     00:00     
 5epel                                | 4.7 kB     00:00     
 6extras                              | 3.4 kB     00:00     
 7updates                             | 3.4 kB     00:00     
 8(1/2): epel/x86_64/updateinfo         | 986 kB   00:00     
 9(2/2): epel/x86_64/primary_db         | 6.7 MB   00:00     
10Resolving Dependencies
11--> Running transaction check
12---> Package mariadb.x86_64 1:5.5.60-1.el7_5 will be installed
13--> Processing Dependency: mariadb-libs(x86-64) = 1:5.5.60-1.el7_5 for package: 1:mariadb-5.5.60-1.el7_5.x86_64
14--> Running transaction check
15---> Package mariadb-libs.x86_64 1:5.5.56-2.el7 will be updated
16---> Package mariadb-libs.x86_64 1:5.5.60-1.el7_5 will be an update
17--> Finished Dependency Resolution
18
19Dependencies Resolved
20
21===========================================================
22 Package        Arch     Version              Repository
23                                                      Size
24===========================================================
25Installing:
26 mariadb        x86_64   1:5.5.60-1.el7_5     base   8.9 M
27Updating for dependencies:
28 mariadb-libs   x86_64   1:5.5.60-1.el7_5     base   758 k
29
30Transaction Summary
31===========================================================
32Install  1 Package
33Upgrade             ( 1 Dependent package)
34
35Total download size: 9.6 M
36Is this ok [y/d/N]: y
37Downloading packages:
38Delta RPMs disabled because /usr/bin/applydeltarpm not installed.
39(1/2): mariadb-libs-5.5.60-1.el7_5.x8 | 758 kB   00:00     
40(2/2): mariadb-5.5.60-1.el7_5.x86_64. | 8.9 MB   00:00     
41-----------------------------------------------------------
42Total                          11 MB/s | 9.6 MB  00:00     
43Running transaction check
44Running transaction test
45Transaction test succeeded
46Running transaction
47  Updating   : 1:mariadb-libs-5.5.60-1.el7_5.x86_64    1/3 
48  Installing : 1:mariadb-5.5.60-1.el7_5.x86_64         2/3 
49  Cleanup    : 1:mariadb-libs-5.5.56-2.el7.x86_64      3/3 
50  Verifying  : 1:mariadb-libs-5.5.60-1.el7_5.x86_64    1/3 
51  Verifying  : 1:mariadb-5.5.60-1.el7_5.x86_64         2/3 
52  Verifying  : 1:mariadb-libs-5.5.56-2.el7.x86_64      3/3 
53
54Installed:
55  mariadb.x86_64 1:5.5.60-1.el7_5                          
56
57Dependency Updated:
58  mariadb-libs.x86_64 1:5.5.60-1.el7_5                     
59
60Complete!

安裝後執行MySQL命令，測試一下是否能夠連接數據庫，客戶端訪問命令格式是MySQL -h主機地址 -u用戶名 －p用戶密碼主機地址使用MySQL數據庫的外部域名。從鏈接數據庫到建立表單的詳細執行過程以下：

一、驗證是否能夠正常鏈接數據庫

1[root@CentOS ~]# mysql -h mysql-cn-north-1-aed0e558da5e4877.public.jcloud.com -P3306 -umysqlxxx –pPasswordxxx

如能夠正常鏈接，能夠新開一個窗口建立SQL腳本用於數據庫的建立和測試數據插入操做，也能夠提早製做好並上傳到客戶端。

建立數據表格，建立一個測試數據庫和測試表便於測試。選擇合適目錄建立SQL腳本文件，能夠用vi ctable.sql 建立，也能夠用其餘文本編輯工具製做，腳本內容以下：

 1[root@CentOS ~]# cat ctable.sql 
 2USE testdb;
 3DROP TABLE IF EXISTS `sqltest`;
 4CREATE TABLE `sqltest` (
 5    `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
 6    `user_id` varchar(20) NOT NULL DEFAULT '',
 7    `vote_num` int(10) unsigned NOT NULL DEFAULT '0',
 8    `group_id` int(10) unsigned NOT NULL DEFAULT '0',
 9    `status` tinyint(2) unsigned NOT NULL DEFAULT '1',
10    `create_time` datetime NOT NULL DEFAULT '0000-00-00 00:00:00',
11    PRIMARY KEY (`id`),
12    KEY `index_user_id` (`user_id`) USING HASH
13) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

二、建立臨時數據產生腳本文件，vi adddb.sql

 1[root@CentOS ~]# cat adddb.sql
 2DELIMITER // -- 修改MySQL delimiter：'//'
 3DROP FUNCTION IF EXISTS `rand_string` //
 4SET NAMES utf8 //
 5CREATE FUNCTION `rand_string` (n INT) RETURNS VARCHAR(255) CHARSET 'utf8'
 6BEGIN 
 7    DECLARE char_str varchar(100) DEFAULT 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789';
 8    DECLARE return_str varchar(255) DEFAULT '';
 9    DECLARE i INT DEFAULT 0;
10    WHILE i < n DO
11        SET return_str = concat(return_str, substring(char_str, FLOOR(1 + RAND()*62), 1));
12        SET i = i+1;
13    END WHILE;
14    RETURN return_str;
15END //

-- 建立插入數據的存儲過程

 1DROP PROCEDURE IF EXISTS `adddb` //
 2CREATE PROCEDURE `adddb`(IN n INT)
 3BEGIN
 4    DECLARE i INT DEFAULT 1;
 5    DECLARE vote_num INT DEFAULT 0;
 6    DECLARE group_id INT DEFAULT 0;
 7    DECLARE status TINYINT DEFAULT 1;
 8    WHILE i < n DO
 9        SET vote_num = FLOOR(1 + RAND() * 10000);
10        SET group_id = FLOOR(0 + RAND()*3);
11        SET status = FLOOR(1 + RAND()*2);
12        INSERT INTO `sqltest` VALUES (NULL, rand_string(20), vote_num, group_id, status, NOW());
13        SET i = i + 1;
14    END WHILE;
15END //DELIMITER ; 

-- 改回默認的 MySQL delimiter：';'

三、登錄數據庫執行ctable.sql和adddb.sql腳本

執行命令：

1MySQL [testdb]> source /root/ctable.sql;
2MySQL [testdb]> source /root/adddb.sql;

增長100條數據

1MySQL [testdb]> call adddb(100);

經過調整adddb（要增長數字）參數的數字，也能夠增長1000條如adddb（1000）。

至此咱們有了數據源的測試環境，接下來能夠開始利用數據工廠進行數據同步。

2、利用數據工廠進行源數據採集

選擇大數據與分析的數據工廠菜單。在鏈接管理中添加鏈接，以下：

創建鏈接時建議點擊鏈接測試按鈕先進行測試數據庫鏈接，不能鏈接時請檢查域名、端口、用戶名密碼是否正確，數據庫是否容許外網訪問。

創建好數據庫鏈接後，就能夠進行數據同步工做了。數據同步工做能夠在數據同步中單獨創建任務設置，也能夠在工做流中經過數據集成選項進行設置。數據集成設置後會自動生成數據同步任務。

調度策略能夠選擇手工執行、週期調度和單次運行三種模式，也能夠直接選擇單次運行。

執行完畢後在數據計算服務中就同步產生了數據。

上述操做能夠用工做流的形式實現，銜接更爲複雜的Spark計算腳本。

成功執行後能夠在運維中心查看執行狀況，在實例列表中的畫圖試布中看到執行節點變爲綠色。

經過以上創建數據同步任務和工做流兩種形式都能實現數據源的數據獲取，數據獲取後就能夠直接使用大數據服務進行數據處理了。在大數據與分析菜單下選擇數據計算服務管理。默認用本身用戶名/PIN（本例用戶名是jdc-14）爲實例名，創建了Default HIVE INSTANCE。

數據的庫表管理下能夠看到剛剛新建的庫MySQLdb和表SQLTest，點擊進入SQLTest表名能夠看到更詳細的表信息如圖。

能夠基於獲取的大數據信息在數據計算服務中進行任務開發，任務開發可使用SQL或開發腳本對數據進行計算。

3、DCS大數據導出

能夠利用這個能力把數據工廠看成簡單DTS工具，把數據傳給目的數據庫。本例在京東雲創建一個MySQL目的數據庫，把SQLTest同步給目的數據庫，實現兩個數據庫的數據同步。

準備好或新創建目的MySQL數據庫 destmysqldb，將數據計算服務的MySQLdb同步給目的數據庫destmysqldb。

在數據工廠菜單下，選擇鏈接管理，新建到destmysqldb的鏈接。

新建同步任務，將數據計算服務的數據同步給destmysqldb數據庫，任務名稱synctodest。選擇數據源端要選擇大數據的數據計算服務，數據庫名爲mysqldb數據表名爲sqltest，表的數據能夠預覽，避免出錯。

在傳輸數據前要在數據庫中事先創建空表結構，執行文章開頭的ctable.sql創建表SQLTest，數據插入時要選擇目的表名。

執行完畢後能夠在destmysqldb中確認結果，經過select count（*） from sqltest 能夠確認數據已經成功導入。

能夠確認利用數據工廠做爲簡單DTS工具進行源數據庫數據同步到目的數據庫，實戰成功！

閱讀原文