通过DataX导入RDS数据

本文主要介绍如何使用阿里巴巴的开源工具DataX实现从关系型数据库到时序引擎的数据迁移。

背景信息

DataX是阿里巴巴集团内被广泛使用的离线数据同步工具,实现包括MySQL、Oracle、SQL Server、PostgreSQL、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、DRDS等各种异构数据源之间高效的数据同步功能。DataX相关使用介绍请参阅DataX的README文档。

下面将首先介绍DataX工具本身,以及以MySQL为例,实现从MySQL迁移到Lindorm时序引擎。本次迁移工作涉及到的两个插件(MySQL ReaderTSDB Writer)。

MySQL Reader

MySQL Reader是DataX的一个插件,实现了从MySQL读取数据。

TSDB Writer

TSDB Writer是DataX的一个插件,实现了将数据点写入到Lindorm数据库时序引擎中。

注意事项

确保与时序引擎的网络是连通的

因为TSDB Writer写入数据的方式是调用HTTP接口(/api/mput以及/api/put)来完成的,所以需要确保迁移任务的进程能正常访问到时序引擎暴露出来的HTTP接口的。否则,会报错Connect Exception异常。

确保与MySQL的网络是连通的

因为MySQL Reader读取数据是通过JDBC来完成的,所以需要确保迁移任务的进程能正常访问到MySQL暴露出来的JDBC接口的。否则,会报错Connect Exception异常。

准备工作

  1. 环境准备。

    • Linux

    • JDK(1.8及以上)

    • Python(推荐Python 2.6.x)

    • MySQL(目前只支持兼容5.x,其他版本暂不兼容)

    • Lindorm时序引擎

  2. 下载DataX及其插件。

    单击链接进行下载。

  3. 利用DataX自带的测试迁移任务,检查迁移流程能否走通。

    这里,我们先以最简单的Stream Reader到Stream Writer为例。因为这两个插件不依赖任何的外部环境,特别适合用来测试流程是否能走通。简单介绍下这两个插件,其中Stream Reader会随机地产生字符串,而Stream Writer则会将接受到的字符串输出到控制台,以此模拟一个最简单的数据迁移过程。

工具部署

将下载后的安装包,解压至某个目录(作为DATAX_HOME),即可运行测试迁移任务:

$ cd ${DATAX_HOME}
$ python bin/datax.py job/job.json

检查任务是否成功

下面是任务结束后的汇总信息,看到如下内容,即可表示迁移任务已成功完成:

任务启动时刻                    : 2019-04-26 11:18:07
任务结束时刻                    : 2019-04-26 11:18:17
任务总计耗时                    :                 10s
任务平均流量                    :          253.91KB/s
记录写入速度                    :          10000rec/s
读出记录总数                    :              100000
读写失败总数                    :                   0
            

配置和启动MySQL到时序引擎的迁移任务

通过上述Stream Reader到Stream Writer的迁移任务,我们就可以确保整个DataX流程是没有问题的。下面就可以真正开始MySQL Reader到TSDB Writer迁移任务了。

参数说明

接下来,我们先看下各个配置项的含义:

MySQL Reader相关

名称

类型

是否必选

描述

默认值

举例

jdbcUrl

String

数据库的JDBC连接信息

jdbc:mysql://127.0.0.1:3306/datax

username

String

数据源的用户名

root

password

String

数据源指定用户名的密码

root

table

String

需要同步的表

book

column

Array

表中需要同步的列名集合

[]

["m"]

splitPk

String

如果指定splitPk,表示用户希望使用splitPk代表的字段进行数据分片

id

TSDB Writer相关

名称

类型

是否必选

描述

默认值

举例

sourceDbType

String

目前支持TSDB和RDB两个取值。其中,时序引擎指OpenTSDB、InfluxDB、Prometheus、TimeScale等;而RDB指MySQL、Oracle、PostgreSQL、DRDS 等

TSDB

RDB

column

Array

表中需要同步的列名集合

[]

["name","type","create_time","price"]

columnType

Array

关系型数据库中表字段,映射到时序引擎中的类型。支持的类型:timestamp:该字段是个时间戳;tag:该字段是个 tag;metric_num:该metric的value 数值类型;metric_string:该metric的value是字符串类型

[]

["tag","tag","timestamp","metric_num"]

endpoint

String

时序引擎的IP:PORT地址

ld-xxxx:8242

multiField

Bool

使用多值方式写入,推荐true

false

true

table

String

导入TSDB的表名

table

username

String

数据库用户名

tsdbuser

password

String

数据库密码

tsdbpsw

ignoreWriteError

Bool

忽略写入错误

false

true

database

String

导入TSDB的数据库名

default

default

配置迁移任务

配置一个从MySQL数据库同步抽取数据到时序引擎的任务,命名为mysql2tsdb.json,完整的配置信息如下(针对各个配置参数的详细说明,请看"参数说明"):

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "column": [
              "name",
              "type",
              "create_time",
              "price",
              "description",
              "available"
            ],
            "connection": [
              {
                "jdbcUrl": [
                  "jdbc:mysql://127.0.0.1:3306/datax"
                ],
                "table": [
                  "book"
                ]
              }
            ],
            "password": "xxx",
            "splitPk": "id",
            "username": "root"
          }
        },
        "writer": {
          "name": "tsdbwriter",
          "parameter": {
            "column": [
              "name",
              "type",
              "create_time",
              "price",
              "description",
              "available"
            ],
            "columnType": [
              "tag",
              "tag",
              "timestamp",
              "field_double",
              "field_string",
              "field_bool"
            ],
            "sourceDbType": "RDB",
            "endpoint": "http://localhost:8242",
            "table": "my_book",
            "multiField": "true", 
            "username":"xxx",
            "password":"xxx",
            "ignoreWriteError":"false",
            "database":"default"
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 1
      }
    }
  }
}          

需要column字段顺序保持一致

需要保证SDB Writer插件中的column字段顺序和MySQL Reader插件中配置的column字段顺序保持一致。否则,数据会错乱。

启动MySQL到时序引擎迁移任务

$ cd ${DATAX_HOME}/..
$ ls
  datax/  datax.tar.gz  mysql2tsdb.json
$ python datax/bin/datax.py mysql2tsdb.json
            

检查任务是否成功

下面是任务结束后的汇总信息,看到如下内容,即可表示迁移任务已成功完成:

任务启动时刻                    : 2019-05-21 18:25:16
任务结束时刻                    : 2019-05-21 18:25:27
任务总计耗时                    :                 11s
任务平均流量                    :                3B/s
记录写入速度                    :              0rec/s
读出记录总数                    :                   3
读写失败总数                    :                   0
            

FAQ

Q:是否支持调整迁移进程的 JVM 内存大小?

A:支持的。以从MySQL到时序引擎的数据迁移任务为例,启动命令如下:

python datax/bin/datax.py mysql2tsdb.json -j "-Xms4096m -Xmx4096m"    

Q:如何给时序引擎设置网络白名单?

A:参考设置白名单

Q:将迁移任务运行在ECS上的用户,如何配置 VPC,以及常见问题有哪些?

A:参考ECS安全组配置案例VPC 常见问题