一. 快速介绍
MysqlWriter 插件实现了写入数据到 Mysql 主库的目的表的功能。在底层实现上， MysqlWriter 通过 JDBC 连接远程 Mysql 数据库，并执行相应的 insert into … 或者 ( replace into …) 的 sql 语句将数据写入 Mysql，内部会分批次提交入库，需要数据库本身采用 innodb 引擎。
MysqlWriter 面向ETL开发工程师，他们使用 MysqlWriter 从数仓导入数据到 Mysql。同时 MysqlWriter 亦可以作为数据迁移工具为DBA等用户提供服务。
二. 实现原理
MysqlWriter 通过 DataX 框架获取 Reader 生成的协议数据，根据你配置的 writeMode 生成
insert into...(当主键/唯一性索引冲突时会写不进去冲突的行)
或者
replace into...(没有遇到主键/唯一性索引冲突时，与 insert into 行为一致，冲突时会用新行替换原有行所有字段) 的语句写入数据到 Mysql。出于性能考虑，采用了PreparedStatement + Batch，并且设置了：rewriteBatchedStatements=true，将数据缓冲到线程上下文 Buffer 中，当 Buffer 累计到预定阈值时，才发起写入请求。
注意：
目的表所在数据库必须是主库才能写入数据；整个任务至少需要具备 insert/replace into…的权限，是否需要其他权限，取决于你任务配置中在 preSql 和 postSql 中指定的语句。
三. 功能说明
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 参数说明
3.2.1 jdbcUrl
-
描述：目的数据库的 JDBC 连接信息。作业运行时，DataX 会在你提供的 jdbcUrl 后面追加如下属性：yearIsDateType=false&zeroDateTimeBehavior=convertToNull&rewriteBatchedStatements=true
注意：
1、在一个数据库上只能配置一个 jdbcUrl 值。这与 MysqlReader 支持多个备库探测不同，因为此处不支持同一个数据库存在多个主库的情况(双主导入数据情况)
2、jdbcUrl按照Mysql官方规范，并可以填写连接附加控制信息，比如想指定连接编码为 gbk ，则在 jdbcUrl 后面追加属性 useUnicode=true&characterEncoding=gbk。具体请参看 Mysql官方文档或者咨询对应 DBA。 -
必选：是
-
默认值：无
3.2.2 username
-
描述：目的数据库的用户名
-
必选：是
-
默认值：无
3.2.3 password
-
描述：目的数据库的密码
-
必选：是
-
默认值：无
3.2.4 table
-
描述：目的表的表名称。支持写入一个或者多个表。当配置为多张表时，必须确保所有表结构保持一致。
注意：table 和 jdbcUrl 必须包含在 connection 配置单元中
-
必选：是
-
默认值：无
3.2.5 column
-
描述：目的表需要写入数据的字段,字段之间用英文逗号分隔。例如: “column”: [“id”,”name”,”age”]。如果要依次写入全部列，使用*表示, 例如: “column”: [“*”]。
column配置项必须指定，不能留空！
注意：
1、我们强烈不推荐你这样配置，因为当你目的表字段个数、类型等有改动时，你的任务可能运行不正确或者失败
2、 column 不能配置任何常量值 -
必选：是
-
默认值：否
3.2.6 session
-
描述: DataX在获取Mysql连接时，执行session指定的SQL语句，修改当前connection session属性
-
必须: 否
-
默认值: 空
3.2.7 preSql
-
描述：写入数据到目的表前，会先执行这里的标准语句。如果 Sql 中有你需要操作到的表名称，请使用 @table 表示，这样在实际执行 Sql 语句时，会对变量按照实际表名称进行替换。比如你的任务是要写入到目的端的100个同构分表(表名称为:datax_00,datax01, … datax_98,datax_99)，并且你希望导入数据前，先对表中数据进行删除操作，那么你可以这样配置：”preSql”:[“delete from 表名”]，效果是：在执行到每个表写入数据前，会先执行对应的 delete from 对应表名称
-
必选：否
-
默认值：无
3.2.8 postSql
-
描述：写入数据到目的表后，会执行这里的标准语句。（原理同 preSql ）
-
必选：否
-
默认值：无
3.2.9 writeMode
-
描述：控制写入数据到目标表采用 insert into 或者 replace into 或者 ON DUPLICATE KEY UPDATE 语句
-
必选：是
所有选项：insert/replace/update -
默认值：insert
3.2.10 batchSize
-
描述：一次性批量提交的记录数大小，该值可以极大减少DataX与Mysql的网络交互次数，并提升整体吞吐量。但是该值设置过大可能会造成DataX运行进程OOM情况。
-
必选：否
-
默认值：1024
3.3 类型转换
类似 MysqlReader ，目前 MysqlWriter 支持大部分 Mysql 类型，但也存在部分个别类型没有支持的情况，请注意检查你的类型。
下面列出 MysqlWriter 针对 Mysql 类型转换列表:
四. 测试案例
现在我们来测试一个从csv读取，然后再写入csv的案例.
4.1 数据准备
4.2 json文件准备
{
"job": {
"setting": {
"speed": {
"channel": 1
}
},
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"username": "root",
"password": "abc123",
"column": [
"id",
"sale_date",
"prod_name",
"sale_nums",
"reserverd1"
],
"splitPk": "id",
"connection": [
{
"table": [
"fact_sale"
],
"jdbcUrl": [
"jdbc:mysql://10.31.1.122:3306/test"
]
}
]
}
},
"writer": {
"name": "mysqlwriter",
"parameter": {
"writeMode": "insert",
"username": "root",
"password": "abc123",
"column": [
"id",
"sale_date",
"prod_name",
"sale_nums",
"reserverd1"
],
"session": [
"set session sql_mode='ANSI'"
],
"preSql": [
"delete from fact_sale_20211123"
],
"connection": [
{
"jdbcUrl": "jdbc:mysql://10.31.1.122:3306/test?useUnicode=true&characterEncoding=gbk",
"table": [
"fact_sale_20211123"
]
}
]
}
}
}
]
}
}
4.3 运行脚本
[15:15:08] [root@10-31-1-119 bin]# python datax.py mysqlreader.json
[15:15:08]
[15:15:08] DataX (DATAX-OPENSOURCE-3.0), From Alibaba !
[15:15:08] Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.
[15:15:08]
[15:15:08]
[15:15:09] 2021-11-23 15:15:15.227 [main] INFO VMInfo - VMInfo# operatingSystem class => sun.management.OperatingSystemImpl
[15:15:09] 2021-11-23 15:15:15.233 [main] INFO Engine - the machine info =>
[15:15:09]
[15:15:09] osInfo: Oracle Corporation 1.8 25.242-b08
[15:15:09] jvmInfo: Linux amd64 3.10.0-1127.el7.x86_64
[15:15:09] cpu num: 8
[15:15:09]
[15:15:09] totalPhysicalMemory: -0.00G
[15:15:09] freePhysicalMemory: -0.00G
[15:15:09] maxFileDescriptorCount: -1
[15:15:09] currentOpenFileDescriptorCount: -1
[15:15:09]
[15:15:09] GC Names [PS MarkSweep, PS Scavenge]
[15:15