DataX-MySQL（读写）
DataX操作MySQL
⼀、从MySQL读取
介绍
MysqlReader插件实现了从Mysql读取数据。在底层实现上，MysqlReader通过JDBC连接远程Mysql数据库，并执⾏相应的sql语句将数据从mysql库中SELECT出来。
不同于其他关系型数据库，MysqlReader不⽀持FetchSize.
实现原理
简⽽⾔之，MysqlReader通过JDBC连接器连接到远程的Mysql数据库，并根据⽤户配置的信息⽣成查询SELECT SQL语句，然后发送到远程Mysql数据库，并将该SQL执⾏返回结果使⽤DataX⾃定义的数据类型拼装为抽象的数据集，并传递给下游Writer处理。
对于⽤户配置Table、Column、Where的信息，MysqlReader将其拼接为SQL语句发送到Mysql数据库；对于⽤户配置querySql信息，MysqlReader直接将其发送到Mysql数据库。
json如下
{
"job": {
"setting": {
"speed": {
"channel": 3
},
"errorLimit": {
"record": 0,
"percentage": 0.02
}
},
"content": [{
"reader": {
"name": "mysqlreader",
"parameter": {
"username": "root",
"password": "123456",
"column": [
"id",
"name"
],
"splitPk": "id",
"connection": [{
"table": [
"datax_test"
],
"jdbcUrl": [
"jdbc:mysql://192.168.1.123:3306/test"
]
}]
}
},
"writer": {
"name": "streamwriter",
"parameter": {
"print": true
}
}
}]
}
}
参数说明--jdbcUrl
描述：描述的是到对端数据库的JDBC连接信息，使⽤JSON的数组描述，并⽀持⼀个库填写多个连接地
址。之所以使⽤JSON数组描述连接信息，是因为阿⾥集团内部⽀持多个IP探测，如果配置了多个，MysqlReader可以依次探测ip的可连接性，直到选择⼀个合法的IP。如果全部连接失败，MysqlReader报错。注意，jdbcUrl必须包含在connection配置单元中。对于阿⾥集团外部使⽤情况，JSON数组填写⼀个JDBC连接即可。
jdbcUrl按照Mysql官⽅规范，并可以填写连接附件控制信息。
必选：是
默认值：⽆
--username
描述：数据源的⽤户名
必选：是
默认值：⽆
--password
描述：数据源指定⽤户名的密码
必选：是
默认值：⽆
--table
描述：所选取的需要同步的表。使⽤JSON的数组描述，因此⽀持多张表同时抽取。当配置为多张表时，⽤户⾃⼰需保证多张表是同⼀schema结构，MysqlReader不予检查表是否同⼀逻辑表。注意，table必须包含在connection配置单元中。
必选：是
默认值：⽆
mysql 字符串转数组--column
描述：所配置的表中需要同步的列名集合，使⽤JSON的数组描述字段信息。⽤户使⽤*代表默认使⽤所有列配置，例如['*']。
⽀持列裁剪，即列可以挑选部分列进⾏导出。
⽀持列换序，即列可以不按照表schema信息进⾏导出。
⽀持常量配置，⽤户需要按照Mysql SQL语法格式: ["id", "`table`", "1", "'bazhen.csy'", "null", "to_char(a + 1)", "2.3" , "true"] id为普通列名，`table`为包含保留在的列名，1为整形数字常量，'bazhen.csy'为字符串常量，null为空指针，to_char(a + 1)为表达式，2.3为浮点数，true为布尔值。
必选：是
默认值：⽆
--splitPk
描述：MysqlReader进⾏数据抽取时，如果指定splitPk，表⽰⽤户希望使⽤splitPk代表的字段进⾏数据分⽚，DataX因此会启动并发任务进⾏数据同步，这样可以⼤⼤提供数据同步的效能。
推荐splitPk⽤户使⽤表主键，因为表主键通常情况下⽐较均匀，因此切分出来的分⽚也不容易出现数据热点。
-
- ⽬前splitPk仅⽀持整形数据切分，不⽀持浮点、字符串、⽇期等其他类型。如果⽤户指定其他⾮⽀持类型，MysqlReader将报错！
--如果splitPk不填写，包括不提供splitPk或者splitPk值为空，DataX视作使⽤单通道同步该表数据。
必选：否
默认值：空
--where
描述：筛选条件，MysqlReader根据指定的column、table、where条件拼接SQL，并根据这个SQL进⾏数据抽取。在实际业务场景中，往往会选择当天的数据进⾏同步，可以将where条件指定为gmt_create > $bizdate 。注意：不可以将where条件指定为limit 10，limit不是SQL的合法where⼦句。
where条件可以有效地进⾏业务增量同步。如果不填写where语句，包括不提供where的key或者value，DataX均视作同步全量数据。
必选：否
默认值：⽆
--querySql
描述：在有些业务场景下，where这⼀配置项不⾜以描述所筛选的条件，⽤户可以通过该配置型来⾃定义筛选SQL。当⽤户配置了这⼀项之后，DataX系统就会忽略
table，column这些配置型，直接使⽤这个配置项的内容对数据进⾏筛选，例如需要进⾏多表join后同步数据，使⽤select a,b from table_a join table_b on table_a.id = table_b.id
当⽤户配置querySql时，MysqlReader直接忽略table、column、where条件的配置，querySql优先级⼤于table、column、where选项。
必选：否
默认值：⽆
mysqlreader类型转换表
请注意:
--除上述罗列字段类型外，其他类型均不⽀持。
--tinyint(1) DataX视作为整形。
--year DataX视作为字符串类型
-
-bit DataX属于未定义⾏为。
执⾏
FengZhendeMacBook-Pro:bin FengZhen$ ./datax.py /Users/FengZhen/Desktop/Hadoop/dataX/json/mysql/reader_all.json
DataX (DATAX-OPENSOURCE-3.0), From Alibaba !
Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.
2018-11-1816:22:04.599 [main] INFO VMInfo - VMInfo# operatingSystem class => sun.management.OperatingSystemImpl
2018-11-1816:22:04.612 [main] INFO Engine - the machine info =>
osInfo: Oracle Corporation 1.825.162-b12
jvmInfo: Mac OS X x86_64 10.13.4
cpu num: 4
totalPhysicalMemory: -0.00G
freePhysicalMemory: -0.00G
maxFileDescriptorCount: -1
currentOpenFileDescriptorCount: -1
GC Names [PS MarkSweep, PS Scavenge]
MEMORY_NAME | allocation_size | init_size
PS Eden Space | 256.00MB | 256.00MB
Code Cache | 240.00MB | 2.44MB
Compressed Class Space | 1,024.00MB | 0.00MB
PS Survivor Space | 42.50MB | 42.50MB
PS Old Gen | 683.00MB | 683.00MB
Metaspace | -0.00MB | 0.00MB
2018-11-1816:22:04.638 [main] INFO Engine -
{
"content":[
{
"reader":{
"name":"mysqlreader",
"parameter":{
"column":[
"id",
"name"
]
,
"connection":[
{
"jdbcUrl":[
"jdbc:mysql://192.168.1.123:3306/test"
],
"table":[
"datax_test"
]
}
],
"password":"******",
"splitPk":"id",
"username":"root"
}
},
"writer":{
"name":"streamwriter",
"parameter":{
"print":true
}
}
}
],
"setting":{
"errorLimit":{
"percentage":0.02,
"record":0
},
"speed":{
"channel":3
}
}
}
2018-11-1816:22:04.673 [main] WARN Engine - prioriy set to 0, because NumberFormatException, the value is: null
2018-11-1816:22:04.678 [main] INFO PerfTrace - PerfTrace traceId=job_-1, isEnable=false, priority=0
2018-11-1816:22:04.678 [main] INFO JobContainer - DataX jobContainer starts job.
2018-11-1816:22:04.681 [main] INFO JobContainer - Set jobId = 0
2018-11-1816:22:05.323 [job-0] INFO OriginalConfPretreatmentUtil - Available jdbcUrl:jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true. 2018-11-1816:22:05.478 [job-0] INFO OriginalConfPretreatmentUtil - table:[datax_test] has columns:[id,name].
2018-11-1816:22:05.490 [job-0] INFO JobContainer - jobContainer starts to do prepare ...
2018-11-1816:22:05.491 [job-0] INFO JobContainer - DataX Reader.Job [mysqlreader] do prepare work .
2018-11-1816:22:05.492 [job-0] INFO JobContainer - DataX Writer.Job [streamwriter] do prepare work .
2018-11-1816:22:05.493 [job-0] INFO JobContainer - jobContainer starts to do split ...
2018-11-1816:22:05.493 [job-0] INFO JobContainer - Job set Channel-Number to 3 channels.
2018-11-1816:22:05.618 [job-0] INFO SingleTableSplitUtil - split pk [sql=SELECT MIN(id),MAX(id) FROM datax_test]
2018-11-1816:22:05.665 [job-0] INFO SingleTableSplitUtil - After split(), allQuerySql=[
select id,name from datax_test where (1 <= id AND id < 2)
select id,name from datax_test where (2 <= id AND id < 3)
select id,name from datax_test where (3 <= id AND id < 4)
select id,name from datax_test where (4 <= id AND id <= 5)
select id,name from datax_test where id IS NULL
].
2018-11-1816:22:05.666 [job-0] INFO JobContainer - DataX Reader.Job [mysqlreader] splits to [5] tasks.
2018-11-1816:22:05.667 [job-0] INFO JobContainer - DataX Writer.Job [streamwriter] splits to [5] tasks.
2018-11-1816:22:05.697 [job-0] INFO JobContainer - jobContainer starts to do schedule ...
2018-11-1816:22:05.721 [job-0] INFO JobContainer - Scheduler starts [1] taskGroups.
2018-11-1816:22:05.744 [job-0] INFO JobContainer - Running by standalone Mode.
2018-11-1816:22:05.758 [taskGroup-0] INFO TaskGroupContainer - taskGroupId=[0] start [3] channels for [5] tasks.
2018-11-1816:22:05.765 [taskGroup-0] INFO Channel - Channel set byte_speed_limit to -1, No bps activated.
2018-11-1816:22:05.766 [taskGroup-0] INFO Channel - Channel set record_speed_limit to -1, No tps activated.
2018-11-1816:22:05.790 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] taskId[0] attemptCount[1] is started
2018-11-1816:22:05.795 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] taskId[1] attemptCount[1] is started
2018-11-1816:22:05.796 [0-0-0-reader] INFO CommonRdbmsReader$Task - Begin to read record by Sql: [select id,name from datax_test where (1 <= id AND id < 2)
] jdbcUrl:[jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
2018-11-1816:22:05.796 [0-0-1-reader] INFO CommonRdbmsReader$Task - Begin to read record by Sql: [select id,name from datax_test where (2 <= id AND id < 3)
] jdbcUrl:[jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
2018-11-1816:22:05.820 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] taskId[2] attemptCount[1] is started
2018-11-1816:22:05.821 [0-0-2-reader] INFO CommonRdbmsReader$Task - Begin to read record by Sql: [select id,name from datax_test where (3 <= id AND id < 4)
] jdbcUrl:[jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
2018-11-1816:22:05.981 [0-0-0-reader] INFO CommonRdbmsReader$Task - Finished read record by Sql: [select id,name from datax_test where (1 <= id AND id < 2)
] jdbcUrl:[jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
1 test1
2018-11-1816:22:06.030 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] taskId[0] is successed, used[241]ms
2018-11-1816:22:06.033 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] taskId[3] attemptCount[1] is started
2018-11-1816:22:06.034 [0-0-3-reader] INFO CommonRdbmsReader$Task - Begin to read record by Sql: [select id,name from datax_test where (4 <= id AND id <= 5)
] jdbcUrl:[jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
2018-11-1816:22:06.041 [0-0-2-reader] INFO CommonRdbmsReader$Task - Finished read record by Sql: [select id,name from datax_test where (3 <= id AND id < 4)
] jdbcUrl:[jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
3 test3
2018-11-1816:22:06.137 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] taskId[2] is successed, used[326]ms
2018-11-1816:22:06.139 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] taskId[4] attemptCount[1] is started
2018-11-1816:22:06.139 [0-0-4-reader] INFO CommonRdbmsReader$Task - Begin to read record by Sql: [select id,name from datax_test where id IS NULL
] jdbcUrl:[jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
2018-11-1816:22:06.157 [0-0-1-reader] INFO CommonRdbmsReader$Task - Finished read record by Sql: [select id,name from datax_test where (2 <= id AND id < 3)
] jdbcUrl:[jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
2 test2
2018-11-1816:22:06.243 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] taskId[1] is successed, used[449]ms
2018-11-1816:22:11.295 [0-0-3-reader] INFO CommonRdbmsReader$Task - Finished read record by Sql: [select id,name from datax_test where (4 <= id AND id <= 5)
] jdbcUrl:[jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
4 test4
5 test5
2018-11-1816:22:11.393 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] taskId[3] is successed, used[5360]ms
2018-11-1816:22:15.784 [job-0] INFO StandAloneJobContainerCommunicator - Total 0 records, 0 bytes | Speed 0B/s, 0 records/s | Error 0 records, 0 bytes | All Task WaitWriterTime 0.000s | All Task WaitReaderTime 0.000s | Percentage 2018-11-1816:22:25.166 [0-0-4-reader] INFO CommonRdbmsReader$Task - Finished read record by Sql: [select id,name from datax_test where id IS NULL
] jdbcUrl:[jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=conver
tToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
2018-11-1816:22:25.413 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] taskId[4] is successed, used[19274]ms
2018-11-1816:22:25.417 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] completed it's tasks.
2018-11-1816:22:25.786 [job-0] INFO StandAloneJobContainerCommunicator - Total 5 records, 30 bytes | Speed 3B/s, 0 records/s | Error 0 records, 0 bytes | All Task WaitWriterTime 0.000s | All Task WaitReaderTime 0.000s | Percentage 2018-11-1816:22:25.786 [job-0] INFO AbstractScheduler - Scheduler accomplished all tasks.
2018-11-1816:22:25.787 [job-0] INFO JobContainer - DataX Writer.Job [streamwriter] do post work.
2018-11-1816:22:25.788 [job-0] INFO JobContainer - DataX Reader.Job [mysqlreader] do post work.
2018-11-1816:22:25.788 [job-0] INFO JobContainer - DataX jobId [0] completed successfully.
2018-11-1816:22:25.791 [job-0] INFO HookInvoker - No hook invoked, because base dir not exists or is a file: /Users/FengZhen/Desktop/Hadoop/dataX/datax/hook
2018-11-1816:22:25.796 [job-0] INFO JobContainer -
[total cpu info] =>
averageCpu | maxDeltaCpu | minDeltaCpu
-1.00% | -1.00% | -1.00%
[total gc info] =>
NAME | totalGCCount | maxDeltaGCCount | minDeltaGCCount | totalGCTime | maxDeltaGCTime | minDeltaGCTime
PS MarkSweep | 0 | 0 | 0 | 0.000s | 0.000s | 0.000s
PS Scavenge | 0 | 0 | 0 | 0.000s | 0.000s | 0.000s
2018-11-1816:22:25.797 [job-0] INFO JobContainer - PerfTrace not enable!
2018-11-1816:22:25.798 [job-0] INFO StandAloneJobContainerCommunicator - Total 5 records, 30 bytes | Speed 1B/s, 0 records/s | Error 0 records, 0 bytes | All Task WaitWriterTime 0.000s | All Task W
aitReaderTime 0.000s | Percentage 2018-11-1816:22:25.799 [job-0] INFO JobContainer -
任务启动时刻 : 2018-11-1816:22:04
任务结束时刻 : 2018-11-1816:22:25
任务总计耗时 : 21s
任务平均流量 : 1B/s
记录写⼊速度 : 0rec/s
读出记录总数 : 5
读写失败总数 : 0
在控制台可看到结果输出
⼆、从MySQL按条件读取数据
json如下
{
"job": {
"setting": {
"speed": {
"channel": 1
}
},
"content": [{
"reader": {
"name": "mysqlreader",
"parameter": {
"username": "root",
"password": "123456",
"connection": [{
"querySql": [
"select * from datax_test where id < 3;"
],
"jdbcUrl": [
"jdbc:mysql://bad_ip:3306/database",
"jdbc:mysql://127.0.0.1:bad_port/database",
"jdbc:mysql://192.168.1.123:3306/test"
]
}]
}
},
"writer": {
"name": "streamwriter",
"parameter": {
"print": true,
"encoding": "UTF-8"
}
}
}]
}
}
执⾏
FengZhendeMacBook-Pro:bin FengZhen$ ./datax.py /Users/FengZhen/Desktop/Hadoop/dataX/json/mysql/reader_select.json
DataX (DATAX-OPENSOURCE-3.0), From Alibaba !
Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.
2018-11-1816:31:20.508 [main] INFO VMInfo - VMInfo# operatingSystem class => sun.management.OperatingSystemImpl
2018-11-1816:31:20.521 [main] INFO Engine - the machine info =>
osInfo: Oracle Corporation 1.825.162-b12
jvmInfo: Mac OS X x86_64 10.13.4
cpu num: 4
totalPhysicalMemory: -0.00G
freePhysicalMemory: -0.00G
maxFileDescriptorCount: -1
currentOpenFileDescriptorCount: -1
GC Names [PS MarkSweep, PS Scavenge]
MEMORY_NAME | allocation_size | init_size
PS Eden Space | 256.00MB | 256.00MB
Code Cache | 240.00MB | 2.44MB
Compressed Class Space | 1,024.00MB | 0.00MB
PS Survivor Space | 42.50MB | 42.50MB
PS Old Gen | 683.00MB | 683.00MB
Metaspace | -0.00MB | 0.00MB
2018-11-1816:31:20.557 [main] INFO Engine -
{
"content":[
{
"reader":{
"name":"mysqlreader",
"parameter":{
"connection":[
{
"jdbcUrl":[
"jdbc:mysql://bad_ip:3306/database",
"jdbc:mysql://127.0.0.1:bad_port/database",
"jdbc:mysql://192.168.1.123:3306/test"
],
"querySql":[
"select * from datax_test where id < 3;"
]
}
],
"password":"******",
"username":"root"
}
},
"writer":{
"name":"streamwriter",
"parameter":{
"encoding":"UTF-8",
"print":true
}
}
}
]
,
"setting":{
"speed":{
"channel":1
}
}
}
2018-11-1816:31:20.609 [main] WARN Engine - prioriy set to 0, because NumberFormatException, the value is: null
2018-11-1816:31:20.612 [main] INFO PerfTrace - PerfTrace traceId=job_-1, isEnable=false, priority=0
2018-11-1816:31:20.613 [main] INFO JobContainer - DataX jobContainer starts job.
2018-11-1816:31:20.618 [main] INFO JobContainer - Set jobId = 0
2018-11-1816:31:21.140 [job-0] WARN DBUtil - test connection of [jdbc:mysql://bad_ip:3306/database] failed, for Code:[MYSQLErrCode-02], Description:[数据库服务的IP地址或者Port错误，请检查填写的IP地址和Port或者联系DBA确认IP地址和P The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server..
2018-11-1816:31:21.143 [job-0] WARN DBUtil - test connection of [jdbc:mysql://127.0.0.1:bad_port/database] failed, for Code:[DBUtilErrorCode-10], Description:[连接数据库失败. 请检查您的账号、密码、数据库名称、IP、Port或者向 DBA 寻求帮2018-11-1816:31:21.498 [job-0] INFO OriginalConfPretreatmentUtil - Available jdbcUrl:jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true. 2018-11-1816:31:21.512 [job-0] INFO JobContainer - jobContainer starts to do prepare ...
2018-11-1816:31:21.518 [job-0] INFO JobContainer - DataX Reader.Job [mysqlreader] do prepare work .
2018-11-1816:31:21.520 [job-0] INFO JobContainer - DataX Writer.Job [streamwriter] do prepare work .
2018-11-1816:31:21.521 [job-0] INFO JobContainer - jobContainer starts to do split ...
2018-11-1816:31:21.524 [job-0] INFO JobContainer - Job set Channel-Number to 1 channels.
2018-11-1816:31:21.546 [job-0] INFO JobContainer - DataX Reader.Job [mysqlreader] splits to [1] tasks.
2018-11-1816:31:21.548 [job-0] INFO JobContainer - DataX Writer.Job [streamwriter] splits to [1] tasks.
2018-11-1816:31:21.587 [job-0] INFO JobContainer - jobContainer starts to do schedule ...
2018-11-1816:31:21.592 [job-0] INFO JobContainer - Scheduler starts [1] taskGroups.
2018-11-1816:31:21.597 [job-0] INFO JobContainer - Running by standalone Mode.
2018-11-1816:31:21.629 [taskGroup-0] INFO TaskGroupContainer - taskGroupId=[0] start [1] channels for [1] tasks.
2018-11-1816:31:21.639 [taskGroup-0] INFO Channel - Channel set byte_speed_limit to -1, No bps activated.
2018-11-1816:31:21.639 [taskGroup-0] INFO Channel - Channel set record_speed_limit to -1, No tps activated.
2018-11-1816:31:21.658 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] taskId[0] attemptCount[1] is started
2018-11-1816:31:21.667 [0-0-0-reader] INFO CommonRdbmsReader$Task - Begin to read record by Sql: [select * from datax_test where id < 3;
] jdbcUrl:[jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
2018-11-1816:31:21.814 [0-0-0-reader] INFO CommonRdbmsReader$Task - Finished read record by Sql: [select * from datax_test where id < 3;
] jdbcUrl:[jdbc:mysql://192.168.1.123:3306/test?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
1 test1
2 test2
2018-11-1816:31:21.865 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] taskId[0] is successed, used[211]ms
2018-11-1816:31:21.866 [taskGroup-0] INFO TaskGroupContainer - taskGroup[0] completed it's tasks.
2018-11-1816:31:31.685 [job-0] INFO StandAloneJobContainerCommunicator - Total 2 records, 12 bytes | Speed 1B/s, 0 records/s | Error 0 records, 0 bytes | All Task WaitWriterTime 0.000s | All Task WaitReaderTime 0.000s | Percentage 2018-11-1816:31:31.685 [job-0] INFO AbstractScheduler - Scheduler accomplished all tasks.
2018-11-1816:31:31.686 [job-0] INFO JobContainer - DataX Writer.Job [streamwriter] do post work.
2018-11-1816:31:31.687 [job-0] INFO JobContainer - DataX Reader.Job [mysqlreader] do post work.
2018-11-1816:31:31.687 [job-0] INFO JobContainer - DataX jobId [0] completed successfully.
2018-11-1816:31:31.688 [job-0] INFO HookInvoker - No hook invoked, because base dir not exists or is a file: /Users/FengZhen/Desktop/Hadoop/dataX/datax/hook
2018-11-1816:31:31.693 [job-0] INFO JobContainer -
[total cpu info] =>
averageCpu | maxDeltaCpu | minDeltaCpu
-1.00% | -1.00% | -1.00%
[total gc info] =>
NAME | totalGCCount | maxDeltaGCCount | minDeltaGCCount | totalGCTime | maxDeltaGCTime | minDeltaGCTime
PS MarkSweep | 0 | 0 | 0 | 0.000s | 0.000s | 0.000s
PS Scavenge | 0 | 0 | 0 | 0.000s | 0.000s | 0.000s
2018-11-1816:31:31.693 [job-0] INFO JobContainer - PerfTrace not enable!
2018-11-1816:31:31.694 [job-0] INFO StandAloneJobContainerCommunicator - Total 2 records, 12 bytes | Speed 1B/s, 0 records/s | Error 0 records, 0 bytes | All Task WaitWriterTime 0.000s | All Task WaitReaderTime 0.000s | Percentage 2018-11-1816:31:31.695 [job-0] INFO JobContainer -
任务启动时刻 : 2018-11-1816:31:20
任务结束时刻 : 2018-11-1816:31:31
任务总计耗时 : 11s
任务平均流量 : 1B/s
记录写⼊速度 : 0rec/s
读出记录总数 : 2
读写失败总数 : 0
三、从MySQL读取写⼊MySQL
写⼊MySQL简介
MysqlWriter 插件实现了写⼊数据到 Mysql 主库的⽬的表的功能。在底层实现上， MysqlWriter 通过 JDBC 连接远程 Mysql 数据库，并执⾏相应的 insert into ... 或者 ( replace into
...) 的 sql 语句将数据写⼊ Mysql，内部会分批次提交⼊库，需要数据库本⾝采⽤ innodb 引擎。
实现原理
MysqlWriter 通过 DataX 框架获取 Reader ⽣成的协议数据，根据你配置的 writeMode ⽣成
(当主键/唯⼀性索引冲突时会写不进去冲突的⾏)
或者
(没有遇到主键/唯⼀性索引冲突时，与 insert into ⾏为⼀致，冲突时会⽤新⾏替换原有⾏所有字段) 的语句写⼊数据到 Mysql。出于性能考虑，采⽤
了 PreparedStatement + Batch，并且设置了：rewriteBatchedStatements=true，将数据缓冲到线程上下⽂ Buffer 中，当 Buffer 累计到预定阈值时，才发起写⼊请求。
json如下
{
"job": {
"setting": {
"speed": {
"channel": 3
},
"errorLimit": {
"record": 0,
"percentage": 0.02
}
},
"content": [{
"reader": {
"name": "mysqlreader",
"parameter": {
"username": "root",
"password": "123456",
"column": [
"id",
"name"
],
"splitPk": "id",
"connection": [{
"table": [
"datax_test"
]
,
"jdbcUrl": [
"jdbc:mysql://192.168.1.123:3306/test"
]
}]
}
},
"writer": {
"name": "mysqlwriter",
"parameter": {
"writeMode": "insert",
"username": "root",
"password": "123456",
"column": [
"id",
"name"
],
"session": [
"set session sql_mode='ANSI'"
],
"preSql": [
"delete from datax_target_test"
]
,
"connection": [{
"jdbcUrl": "jdbc:mysql://192.168.1.123:3306/test?useUnicode=true&characterEncoding=gbk",
"table": [
"datax_target_test"
]
}]
}
}
}]
}
}
执⾏
FengZhendeMacBook-Pro:bin FengZhen$ ./datax.py /Users/FengZhen/Desktop/Hadoop/dataX/json/sql2mysql.json
DataX (DATAX-OPENSOURCE-3.0), From Alibaba !
Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.
2018-11-1816:49:13.176 [main] INFO VMInfo - VMInfo# operatingSystem class => sun.management.OperatingSystemImpl
2018-11-1816:49:13.189 [main] INFO Engine - the machine info =>
osInfo: Oracle Corporation 1.825.162-b12
jvmInfo: Mac OS X x86_64 10.13.4
cpu num: 4
totalPhysicalMemory: -0.00G
freePhysicalMemory: -0.00G
maxFileDescriptorCount: -1
currentOpenFileDescriptorCount: -1
GC Names [PS MarkSweep, PS Scavenge]
MEMORY_NAME | allocation_size | init_size
PS Eden Space | 256.00MB | 256.00MB
Code Cache | 240.00MB | 2.44MB
版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。
发表评论