DataX使用记录

2019-12-18 11:29:55 浏览数 (1)

说明

本文描述问题及解决方法同样适用于 腾讯云Snova云数仓。

数据导入介绍

Greenplum(以下简称GP)支持多种数据导入方法,比如GP自带的gpfdist,通过gpfdist 外部表的形式将远端服务器上的数据并行导入到GP中,再比如GP自带的COPY命令,能够将本地的数据按照一定格式导入到GP中。除此之外,还有一些比较优秀的第三方导入工具,本文主要介绍DataX。

DataX概述

DataX是一款能够完成异构数据源之间数据迁移的软件,DataX采用FrameWork Plugin的软件架构,扩展方便。所有数据源中的数据都先转换为DataX的格式,然后在转换成目的端的数据格式,避免出现各异构数据源之间的类型相互转换。

image.pngimage.png

通过DataX,可以容易得将现有数据从mysql、sqlserver、oracle等迁移到Snova中。

支持的数据类型

使用DataX进行数据导入时,第一步是将源端数据源的数据转换为DataX的数据类型,然后将DataX的数据类型转换为目标数据源的数据类型。因此,在使用DataX前,需要先确认是否存在DataX不支持的数据类型,现有数据源中的数据类型与DataX的类型映射如下:

Greenplum

DataX数据类型

GP数据类型

Long

bigint, bigserial, integer, smallint, serial

Double

double precision, money, numeric, real

String

varchar, char, text, bit

Date

date, time, timestamp

Boolean

bool

Bytes

bytea

Mysql

DataX数据类型

Mysql数据类型

Long

int, tinyint, smallint, mediumint, int, bigint

Double

float, double, decimal

String

varchar, char, tinytext, text, mediumtext, longtext, year

Date

date, datetime, timestamp, time

Boolean

bit, bool

Bytes

tinyblob, mediumblob, blob, longblob, varbinary

Oracle

DataX数据类型

Oracle数据类型

Long

NUMBER,INTEGER,INT,SMALLINT

Double

NUMERIC,DECIMAL,FLOAT,DOUBLE PRECISION,REAL

String

char,nchar,ntext,nvarchar,text,varchar,nvarchar(MAX),varchar(MAX)

Date

LONG,CHAR,NCHAR,VARCHAR,VARCHAR2,NVARCHAR2,CLOB,NCLOB,CHARACTER,CHARACTER VARYING,CHAR VARYING,NATIONAL CHARACTER,NATIONAL CHAR,NATIONAL CHARACTER VARYING,NATIONAL CHAR VARYING,NCHAR VARYING

Boolean

bit, bool

Bytes

BLOB,BFILE,RAW,LONG RAW

使用方法

本节将会介绍把Mysql中的数据导入到GP中的方法。

工具安装

DataX的安装可以参照 github。

数据准备

Mysql创建表

在mysql的数据库db1中创建test1表。

代码语言:txt复制
create table test2 (id int auto_increment primary key not null, ca int, cb int, cc varchar(50));

Mysql数据生成

向表test1中插入测试数据1000条。

首先创建插入的存储过程:

代码语言:txt复制
DELIMITER //
CREATE PROCEDURE db1.insert_multi_records(IN n INT)
BEGIN
    DECLARE i INT DEFAULT 1;
    DECLARE va  INT DEFAULT 0;
    DECLARE vb INT DEFAULT 0;
    DECLARE status TINYINT DEFAULT 1;
    WHILE i < n DO
        SET va= FLOOR(1   RAND() * 10000);
        SET vb = FLOOR(0   RAND()*3);
        INSERT INTO db1.test2 VALUES (NULL, va, vb, 'abc');
        SET i = i   1;
    END WHILE;
END //
DELIMITER ;

插入1000条数据:

代码语言:txt复制
CALL db1.insert_multi_records(1000);

GP创建表

在GP中创建与Mysql结构一致的表,当然GP中的表可以与Mysql中不一致,在进行导入作业时,可以通过参数设置那些列需要被导入,参考参数解释。

代码语言:txt复制
CREATE TABLE test2(id SERIAL NOT NULL,ca INTEGER, cb INTEGER,cc VARCHAR(50)) DISTRIBUTED BY  (id);

数据导入

由于GP的底层存储引擎使用的是postgresql,因此,写入端的writer直接选择postgresqlwriter即可。从Mysql将数据导入到GP的配置文件如下:

代码语言:txt复制
{
    "job": {
        "setting": {
            "speed": {
		    "channel": 3,
		    "byte": 1048576,
		    "record": 1000
            },
            "errorLimit": {
                "record": 2,
                "percentage": 0.02
            }
        },
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "username": "****",
                        "password": "****",
                        "column": [
                            "*"
                        ],
		            	"splitPk" : "id",
                        "connection": [
                            {
                                "table": [
                                    "test1"
                                ],
                                "jdbcUrl": [
									"jdbc:mysql://***:***/db1"
                                ]
                            }
                        ]
                    }
                },
               "writer": {
                    "name": "postgresqlwriter",
                    "parameter": {
                        "username": "******",
                        "password": "******",
                        "column": [
                            "*"
                        ],
                        "preSql": [
                            "delete from test1"
                        ],
                        "postSql":[
					     "select count(*) from test2"
				     	],
					"batchsize":100,
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:postgresql://****:**/db1",
                                "table": [
                                    "test1"
                                ]
                            }
                        ],
			            "encoding":"utf-8"
                    }
                }
            }
        ]
    }
}

参数解释

连接信息

reader

所有与读取插件相关的参数都在job.content.reader中,以下参数省略所有job.content.reader前缀,例如 name等价于job.content.reader.name

  1. name表示插件的类型,在我们的例子中设置为mysqlreader,表示源端数据源类型为mysql。
  2. parameter表示所有与连接url相关的配置。 2.1 username与password分别表示连接数据库是的用户名和密码。 2.2 connection中的为jdbcUrl为连接数据库时的Url,各数据源的连接url有细微区别,具体可参考:
代码语言:txt复制
mysql:http://dev.mysql.com/doc/connector-j/en/connector-j-reference-configuration-properties.html
代码语言:txt复制
sqlserver:http://technet.microsoft.com/zh-cn/library/ms378749(v=SQL.110).aspx
代码语言:txt复制
oracle:http://www.oracle.com/technetwork/database/enterprise-edition/documentation/index.html
writer

与写插件相关的参数都在job.content.writer中,writer中与连接相关的参数与reader中的各参数意义相同,只是需要设置为目的数据源的相关配置。例如在我们的例子中,使用postgresql连接GP,因此name设置为postgresqlwriter。

并发控制

在DataX中,可以将一个作业根据策略划分成多个Task,Task作为迁移过程中执行的最小作业单位等待调度。

多表导入

数据导入时,可以对单张表或者多张表进行导入,在我们的例子中,只对单张表进行了导入,实际上可以通过对reader.connection.table进行设置,例如例子中的reader.connection参数设置以下参数,表示多张表同时导入。但是,多张表进行导入时,多张表的结果必须完全一致

代码语言:txt复制
"connection": [{
						"table": [
							"test2",
                            "test3"
						],
						"jdbcUrl": [
							"jdbc:mysql://172.16.0.14:3306/db1"
						]
					}]
单表导入

出了使用多表进行并发,如果导入时只有一张表,同样可以进行并发导入。例如例子中的以下配置可以对表进行划分成多个Task。

代码语言:txt复制
"reader": {
					"splitPk": "id"
			}

splitPk用设置对表进行划分时依赖的列,不一定要设置为主键,可以设置为能将表均匀散列的列,如果splitPk选取不合适,会出现Task负载不均的情况,并发效果较差。另外,splitPk只能选取数据类型为整形或者字符串类型的列,其他类型会报错。

例如,对于例子中的配置而言,以id作为splitPk作为划分键时,假设数据库中共计50W条数据,id为自增长的整型,并且取值为1~500000,通过拆分以后,会划分成50个数据量为1W的任务,并发导入。

并发通道

在job.setting.speed中,同样存在控制并发导入的参数,channel用来控制并发通道的数量。即,即使通过上述的splitPk对表进行了划分,但是如果channel设置不合理,也不能并发导入,比如,如果channel设置为1,那么50任务会串行通过该唯一通道进行迁移,而如果channel设置为10,会启动10个channel同时进行迁移。因此channel与splitPk必须搭配使用才能生效。

批量控制

在writer中,可以设置batchsize对插入进行批量处理。例如,不设置batchsize时,如果要插入两条数据,DataX会通过以下语句进行:

代码语言:txt复制
insert into test2 (id, ca, cb, cc) values(1, 1, 2, 'abcdefg');
insert into test2 (id, ca, cb, cc) values(2, 1, 2, 'gfedcba');

这会导致DataX与GP之间会通过两次实务来插入量条数据,增加通信成本,而如果设置了batchsize,比如讲batchsize设置为100时,DataX会通过以下语句进行:

代码语言:txt复制
insert into test2 (id, ca, cb, cc) values(1, 1, 2, 'abcdefg'),values(2, 1, 2, 'gfedcba');

当batchsize设置在100左右时,能够大幅度降低DataX和目标数据源之间的通信成本,提升导入性能。

条件过滤

结构过滤

通常情况下,源端数据源的表结构与目标数据源的表结构,如例子中的mysql与GP中数据迁移前后的表列数与类型完全一致,但是在某些情况下,迁移前后的结构是不一致的,通常表现后目标数据库中的表中只保留源端数据库表中的某几列,这时,可以通过reader.parameter和writer.parameter中的column参数来指定需要进行导入的列。例如:

代码语言:txt复制
"reader": {
					"column": [
						"id",
                        "ca",
                        "cc"
					],
					"splitPk": "id",
			}
代码语言:txt复制
"writer": {
					"column": [
						"id",
                        "ca",
                        "cc"
					]
			},

如上配置,从mysql导入到GP中以后,只保留id,ca,cc三列数据,cb列的数据被丢弃。

数据过滤

例子中的配置,在每次作业时,会将源端数据库表中的数据全量导入到目标数据库表中,DataX支持对reader进行where条件配置,对源端数据库表的数据进行过滤,例如:

代码语言:txt复制
"reader": {
				"where" : "id>100"
			}

通过配置where语句,通常可以用来对表的数据进行增量导入,例如,若表中有一个时间戳字段,下次导入时,可以选择一上次导入的最后一个时间戳开始,完成增量导入。

数据清理

在DataX进行作业前后,可以分别配置sql语句进行额外的处理,例如preSql会在导入任务执行前执行,postSql会在导入完成后执行。

如例子中的配置,每次在将数据导入到test2表中前,先将test2中的数据删除,避免重复导入。

代码语言:txt复制
"reader": {
				"preSql" : [
                    "delete from test2"
                ]
			}

总结

虽然DataX提供了丰富的从其他数据库迁移到GP的方法,但是将数据导入到GP时,所有数据需要经过master根据分布键计算后再次进行分发,master的会成为数据导入过程中的性能瓶颈,后续会介绍其他不经过master的更高性能的导入方法。

0 人点赞