DataX MysqlWriter

1 快速介绍

• MysqlWriter插件实现了写入数据到Mysql主库的目的表的功能。在底层实现上， MysqlWriter通过JDBC连接远程Mysql数据库，并执行相应的insert into … 或者（replace into …）的Sql语句将数据写入Mysql，内部会分批次提交入库，需要数据库本身采用innodb引擎。
• MysqlWriter面向ETL开发工程师，他们使用MysqlWriter从数仓导入数据到Mysql。同时MysqlWriter亦可以作为数据迁移工具为DBA等用户提供服务。

2 实现原理

MysqlWriter通过DataX框架获取Reader生成的协议数据，根据用户配置的 writeMode生成。您可以使用

• insert into…当主键/唯一性索引冲突时会写不进去冲突的行

或者

• replace into…没有遇到主键/唯一性索引冲突时，与insert into行为一致，冲突时会用新行替换原有行所有字段。出于性能考虑，采用 PreparedStatement+Batch，并且设置：rewriteBatchedStatements=true，将数据缓冲到线程上下文Buffer中，当 Buffer累计到预定阈值时，才发起写入请求。

3 功能说明

3.1 配置样例

• 这里使用一份从内存产生到Mysql导入的数据。

{
    "job": {
        "setting": {
            "speed": {
                "channel": 1
            }
        },
        "content": [
            {
                 "reader": {
                    "name": "streamreader",
                    "parameter": {
                        "column" : [
                            {
                                "value": "DataX",
                                "type": "string"
                            },
                            {
                                "value": 19880808,
                                "type": "long"
                            },
                            {
                                "value": "1988-08-08 08:08:08",
                                "type": "date"
                            },
                            {
                                "value": true,
                                "type": "bool"
                            },
                            {
                                "value": "test",
                                "type": "bytes"
                            }
                        ],
                        "sliceRecordCount": 1000
                    }
                },
                "writer": {
                    "name": "mysqlwriter",
                    "parameter": {
                        "writeMode": "insert",
                        "username": "root",
                        "password": "root",
                        "column": [
                            "id",
                            "name"
                        ],
                        "session": [
                        	"set session sql_mode='ANSI'"
                        ],
                        "preSql": [
                            "delete from test"
                        ],
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:mysql://127.0.0.1:3306/datax?useUnicode=true&characterEncoding=gbk",
                                "table": [
                                    "test"
                                ]
                            }
                        ]
                    }
                }
            }
        ]
    }
}

3.2 参数说明

• jdbcUrl
• 描述：目的数据库的JDBC连接信息。作业运行时，DataX会在你提供的 jdbcUrl后面追加如下属性：yearIsDateType=false&zeroDateTimeBehavior=convertToNull&rewriteBatchedStatements=true

*注意：

• • 在一个数据库上只能配置一个jdbcUrl值。这与MysqlReader支持多个备库探测不同，因为此处不支持同一个数据库存在多个主库的情况（双主导入数据情况）。
  • jdbcUrl按照Mysql官方规范，并可以填写连接附加控制信息，比如想指定连接编码为gbk ，则在jdbcUrl后面追加属性 useUnicode=true&characterEncoding=gbk。具体请参看Mysql官方文档或者咨询对应DBA。
  • 必选：是
  • 默认值：无
• username
  • 描述：目的数据库的用户名
  • 必选：是
  • 默认值：无
• password
  • 描述：目的数据库的密码
  • 必选：是
  • 默认值：无
• table
  • 描述：目的表的表名称；支持写入一个或者多个表；当配置为多张表时，必须确保所有表结构保持一致。

*注意：table和jdbcUrl必须包含在connection配置单元中

• • 必选：是
  • 默认值：无
• column
  • 描述：目的表需要写入数据的字段，字段之间用英文逗号分隔。例如： “column”: [“id”,”name”,”age”]。如果要依次写入全部列，使用*表示， 例如: “column”: [“*”]。
  • 必选：是
  • 默认值：否
• session
  • 描述: DataX在获取Mysql连接时，执行session指定的SQL语句，修改当前connection session属性
  • 必须: 否
  • 默认值: 空
• preSql
  • 描述：写入数据到目的表前，会先执行这里的标准语句。如果Sql中有你需要操作到的表名称，请使用@table表示，这样在实际执行Sql语句时，会对变量按照实际表名称进行替换。比如你的任务是要写入到目的端的100个同构分表（表名称为:datax_00,datax01, … datax_98,datax_99），并且你希望导入数据前，先对表中数据进行删除操作，那么你可以这样配置：”preSql”:[“delete from 表名”]，效果是：在执行到每个表写入数据前，会先执行对应的delete from对应表名称
  • 必选：否
  • 默认值：无
• postSql
  • 描述：写入数据到目的表后，会执行这里的标准语句。（原理同 preSql 。）
  • 必选：否
  • 默认值：无
• writeMode
  • 描述：控制写入数据到目标表采用insert into或者replace into或者 ON DUPLICATE KEY UPDATE语句
  • 必选：是
  • 所有选项：insert/replace/update
  • 默认值：insert
• batchSize
  • 描述：一次性批量提交的记录数大小，该值可以极大减少DataX与Mysql的网络交互次数，并提升整体吞吐量。但是该值设置过大可能会造成DataX运行进程OOM情况。
  • 必选：否
  • 默认值：1024

3.3 类型转换

类似MysqlReader，目前MysqlWriter支持大部分Mysql类型，但也存在部分个别类型没有支持的情况，请注意检查你的类型。

下面列出MysqlWriter针对Mysql类型转换列表：

*注意：bit类型目前是未定义类型转换。

4 性能报告

4.1 环境准备

4.1.1 数据特征

建表语句：

CREATE TABLE `datax_mysqlwriter_perf_00` (
`biz_order_id` bigint(20) NOT NULL AUTO_INCREMENT  COMMENT 'id',
`key_value` varchar(4000) NOT NULL COMMENT 'Key-value的内容',
`gmt_create` datetime NOT NULL COMMENT '创建时间',
`gmt_modified` datetime NOT NULL COMMENT '修改时间',
`attribute_cc` int(11) DEFAULT NULL COMMENT '防止并发修改的标志',
`value_type` int(11) NOT NULL DEFAULT '0' COMMENT '类型',
`buyer_id` bigint(20) DEFAULT NULL COMMENT 'buyerid',
`seller_id` bigint(20) DEFAULT NULL COMMENT 'seller_id',
PRIMARY KEY (`biz_order_id`,`value_type`),
KEY `idx_biz_vertical_gmtmodified` (`gmt_modified`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='datax perf test'

单行记录类似于：

   key_value: ;orderIds:20148888888,2014888888813800;
  gmt_create: 2011-09-24 11:07:20
gmt_modified: 2011-10-24 17:56:34
attribute_cc: 1
  value_type: 3
	buyer_id: 8888888
   seller_id: 1

4.1.2 机器参数

• 执行DataX的机器参数为:
  • cpu: 24核 Intel(R) Xeon(R) CPU E5-2630 0 @ 2.30GHz
  • mem: 48GB
  • net: 千兆双网卡
  • disc: DataX 数据不落磁盘，不统计此项
• Mysql数据库机器参数为:
  • cpu: 32核 Intel(R) Xeon(R) CPU E5-2650 v2 @ 2.60GHz
  • mem: 256GB
  • net: 千兆双网卡
  • disc: BTWL419303E2800RGN INTEL SSDSC2BB800G4 D2010370

4.1.3 DataX jvm 参数

-Xms1024m -Xmx1024m -XX:+HeapDumpOnOutOfMemoryError

FAQ

Q: MysqlWriter执行postSql语句报错，那么数据导入到目标数据库了吗?

A: DataX 入过程存在三块逻辑，pre操作、导入操作、post 操作，其中任意一环报错，DataX作业报错。由于DataX不能保证在同一个事务完成上述几个操作，因此有可能数据已经落入到目标端。

Q: 按照上述说法，那么有部分脏数据导入数据库，如果影响到线上数据库怎么办?

A: 目前有两种解法，第一种配置pre语句，该sql可以清理当天导入数据，DataX 每次导入时候可以把上次清理干净并导入完整数据。第二种，向临时表导入数据，完成后再rename到线上表。

Q: 上面第二种方法可以避免对线上数据造成影响，那我具体怎样操作?

A: 可以配置临时表导入。