DataX 实战：MySQL 读写与自定义 SQL 同步任务配置详解

1. 引言

DataX 是阿里巴巴开源的数据同步工具，支持多种异构数据源之间的高效数据迁移。本文通过三个实战案例，详细介绍如何使用 DataX 完成 MySQL 数据的读取、写入以及自定义 SQL 同步任务，帮助读者快速上手 DataX 的作业配置。

2. 环境准备

在开始之前，请确保已完成以下准备工作：

• 已安装并配置好 DataX 运行环境，具备 Java 运行环境。
• 本地 MySQL 服务正常运行，并准备好测试数据库和表。
• 准备好 DataX 的作业配置文件，通常为 JSON 格式。

3. 案例一：MySQL 读取到本地（mysqlreader + streamwriter）

第一个案例演示如何从 MySQL 数据库同步抽取数据到本地，使用 mysqlreader 作为读取插件，streamwriter 作为写入插件，将数据打印到控制台。

3.1 作业配置

以下是完整的作业配置脚本：

{
"job": {
"setting": {
"speed": {
"channel": 3
},
"errorLimit": {
"record": 0,
"percentage": 0.02
}
},
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"username": "root",
"password": "root",
"column": [
"id",
"name"
],
"splitPk": "db_id",
"connection": [
{
"table": [
"table"
],
"jdbcUrl": [
"jdbc:mysql://127.0.0.1:3306/database"
]
}
]
}
},
"writer": {
"name": "streamwriter",
"parameter": {
"print": true
}
}
}
]
}
}

3.2 配置说明

下面对上述配置中的关键参数进行说明：

3.3 运行结果

执行该作业后，DataX 会从 MySQL 的 table 表中读取 id 和 name 两列数据，并通过 streamwriter 打印到控制台。由于设置了 3 个并发通道，读取速度会有所提升。

4. 案例二：自定义 SQL 同步任务（mysqlreader + streamwriter）

第二个案例演示如何通过自定义 SQL 语句从 MySQL 同步数据到本地，使用 querySql 参数代替简单的表名读取。

4.1 作业配置

以下是使用自定义 SQL 的作业配置脚本：

{
"job": {
"setting": {
"speed": {
"channel": 1
}
},
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"username": "root",
"password": "root",
"connection": [
{
"querySql": [
"select db_id,on_line_flag from db_info where db_id < 10;"
],
"jdbcUrl": [
"jdbc:mysql://bad_ip:3306/database",
"jdbc:mysql://127.0.0.1:bad_port/database",
"jdbc:mysql://127.0.0.1:3306/database"
]
}
]
}
},
"writer": {
"name": "streamwriter",
"parameter": {
"print": false,
"encoding": "UTF-8"
}
}
}
]
}
}

4.2 配置说明

该配置与案例一的主要区别在于使用了 querySql 参数，直接指定查询语句来读取数据。需要注意以下几点：

• querySql：自定义查询语句，可以灵活筛选需要同步的数据，这里查询 db_id 小于 10 的记录。
• jdbcUrl 多地址：配置了多个 JDBC 地址，DataX 会依次尝试连接，直到成功为止。示例中前两个地址是故意配置的错误地址，用于演示容错机制。
• encoding：streamwriter 的输出编码，设置为 UTF-8。
• print：设置为 false，表示不打印数据到控制台。

4.3 运行结果

执行该作业后，DataX 会通过自定义 SQL 查询 db_info 表中 db_id 小于 10 的记录，并将结果输出。由于配置了多个 JDBC 地址，DataX 会跳过不可用的地址，最终连接到有效的数据库。

5. 案例三：内存数据写入 MySQL（streamreader + mysqlwriter）

第三个案例演示反向操作，即从内存产生数据并写入 MySQL 数据库，使用 streamreader 作为读取插件，mysqlwriter 作为写入插件。

5.1 作业配置

以下是完整的作业配置脚本：

{
"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"
]
}
]
}
}
}
]
}
}

5.2 配置说明

下面对该配置中的关键参数进行说明：

5.3 运行结果

执行该作业后，DataX 会在内存中生成 1000 条测试数据，并在写入前清空 test 表，然后将数据插入到 MySQL 的 test 表中。

6. 总结

本文通过三个实战案例，详细介绍了 DataX 在 MySQL 数据同步中的典型用法：

• MySQL 读取到本地：使用 mysqlreader 和 streamwriter 实现数据抽取与打印。
• 自定义 SQL 同步：通过 querySql 灵活筛选数据，并演示了多 JDBC 地址的容错机制。
• 内存数据写入 MySQL：使用 streamreader 生成测试数据，通过 mysqlwriter 写入数据库。

掌握这些基础配置后，读者可以根据实际业务需求，灵活组合 DataX 的各种插件，构建高效可靠的数据同步管道。