干货 | 快速实现数据导入及简单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工具进行源数据库数据同步到目的数据库，实战成功！

阅读原文