作者：愿许浪尽咫尺 \
链接：https://juejin.cn/post/7077744714954309669
前言
咱们公司有个我的项目的数据量高达五千万，然而因为报表那块数据不太精确，业务库和报表库又是跨库操作，所以并不能应用 SQL 来进行同步。过后的打算是通过 mysqldump 或者存储的形式来进行同步，然而尝试后发现这些计划都不切实际：
mysqldump：不仅备份须要工夫，同步也须要工夫，而且在备份的过程，可能还会有数据产出（也就是说同步等于没同步）存储形式：这个效率太慢了，要是数据量少还好，咱们应用这个形式的时候，三个小时才同步两千条数据 …
前面在网上查看后：
- 发现 DataX 这个工具用来同步不仅速度快，而且同步的数据量基本上也相差无几。
一、DataX 简介
DataX 是阿里云 DataWorks 数据集成 的开源版本，次要就是用于实现数据间的离线同步。 DataX 致力于实现包含关系型数据库（MySQL、Oracle 等）、HDFS、Hive、ODPS、HBase、FTP 等 各种异构数据源（即不同的数据库） 间稳固高效的数据同步性能。
为了 解决异构数据源同步问题，DataX 将简单的网状同步链路变成了星型数据链路，DataX 作为两头传输载体负责连贯各种数据源；当须要接入一个新的数据源时，只须要将此数据源对接到 DataX，便能跟已有的数据源作为无缝数据同步。
1.DataX3.0 框架设计
DataX 采纳 Framework + Plugin 架构，将数据源读取和写入形象称为 Reader/Writer 插件，纳入到整个同步框架中。
| 角色 | 作用 |
|---|---|
| Reader（采集模块） | 负责采集数据源的数据，将数据发送给 Framework。 |
| Writer（写入模块） | 负责一直向 Framework 中取数据，并将数据写入到目标端。 |
| Framework（中间商） | 负责连贯 Reader 和 Writer，作为两者的数据传输通道，并解决缓冲，流控，并发，数据转换等核心技术问题。 |
2.DataX3.0 外围架构
DataX 实现单个数据同步的作业，咱们称为 Job，DataX 接管到一个 Job 后，将启动一个过程来实现整个作业同步过程。DataX Job 模块是单个作业的中枢治理节点，承当了数据清理、子工作切分、TaskGroup 治理等性能。
DataX Job 启动后，会依据不同源端的切分策略，将 Job 切分成多个小的 Task (子工作)，以便于并发执行。接着 DataX Job 会调用 Scheduler 模块，依据配置的并发数量，将拆分成的 Task 重新组合，组装成 TaskGroup（工作组）
每一个 Task 都由 TaskGroup 负责启动，Task 启动后，会固定启动 Reader --> Channel --> Writer 线程来实现工作同步工作。DataX 作业运行启动后，Job 会对 TaskGroup 进行监控操作，期待所有 TaskGroup 实现后，Job 便会胜利退出（异样退出时 值非 0）
DataX 调度过程：
首先 DataX Job 模块会依据分库分表切分成若干个 Task，而后依据用户配置并发数，来计算须要调配多少个 TaskGroup（计算过程：Task / Channel = TaskGroup）最初由 TaskGroup 依据调配好的并发数来运行 Task（工作）
二、应用 DataX 实现数据同步
筹备工作：
- JDK（1.8 以上，举荐 1.8）
- Python（2，3 版本都能够）
- Apache Maven 3.x（Compile DataX）（手动打包应用，应用
tar包形式不须要装置）
| 主机名 | 操作系统 | IP 地址 | 软件包 |
|---|---|---|---|
| MySQL-1 | CentOS 7.4 | 192.168.1.1 | jdk-8u181-linux-x64.tar.gz datax.tar.gz |
| MySQL-2 | CentOS 7.4 | 192.168.1.2 |
装置 JDK：
[root@MySQL-1 ~]# ls
anaconda-ks.cfg jdk-8u181-linux-x64.tar.gz
[root@MySQL-1 ~]# tar zxf jdk-8u181-linux-x64.tar.gz
[root@DataX ~]# ls
anaconda-ks.cfg jdk1.8.0_181 jdk-8u181-linux-x64.tar.gz
[root@MySQL-1 ~]# mv jdk1.8.0_181 /usr/local/java
[root@MySQL-1 ~]# cat <<END >> /etc/profile
export JAVA_HOME=/usr/local/java
export PATH=$PATH:"$JAVA_HOME/bin"
END
[root@MySQL-1 ~]# source /etc/profile
[root@MySQL-1 ~]# java -version因为 CentOS 7 上自带 Python 2.7 的软件包，所以不须要进行装置。
1.Linux 上装置 DataX 软件
[root@MySQL-1 ~]# wget http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz
[root@MySQL-1 ~]# tar zxf datax.tar.gz -C /usr/local/
[root@MySQL-1 ~]# rm -rf /usr/local/datax/plugin/*/._* # 须要删除暗藏文件 (重要)当未删除时，可能会输入：[/usr/local/datax/plugin/reader/._drdsreader/plugin.json] 不存在. 请查看您的配置文件.
验证：
[root@MySQL-1 ~]# cd /usr/local/datax/bin
[root@MySQL-1 ~]# python datax.py ../job/job.json # 用来验证是否装置胜利输入：
2021-12-13 19:26:28.828 [job-0] INFO JobContainer - PerfTrace not enable!
2021-12-13 19:26:28.829 [job-0] INFO StandAloneJobContainerCommunicator - Total 100000 records, 2600000 bytes | Speed 253.91KB/s, 10000 records/s | Error 0 records, 0 bytes | All Task WaitWriterTime 0.060s | All Task WaitReaderTime 0.068s | Percentage 100.00%
2021-12-13 19:26:28.829 [job-0] INFO JobContainer -
工作启动时刻 : 2021-12-13 19:26:18
工作完结时刻 : 2021-12-13 19:26:28
工作总计耗时 : 10s
工作均匀流量 : 253.91KB/s
记录写入速度 : 10000rec/s
读出记录总数 : 100000
读写失败总数 : 0举荐一个开源收费的 Spring Boot 最全教程：
https://github.com/javastacks/spring-boot-best-practice
2.DataX 根本应用
查看 streamreader --> streamwriter 的模板：
[root@MySQL-1 ~]# python /usr/local/datax/bin/datax.py -r streamreader -w streamwriter输入：
DataX (DATAX-OPENSOURCE-3.0), From Alibaba !
Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.
Please refer to the streamreader document:
https://github.com/alibaba/DataX/blob/master/streamreader/doc/streamreader.md
Please refer to the streamwriter document:
https://github.com/alibaba/DataX/blob/master/streamwriter/doc/streamwriter.md
Please save the following configuration as a json file and use
python {DATAX_HOME}/bin/datax.py {JSON_FILE_NAME}.json
to run the job.
{
"job": {
"content": [
{
"reader": {
"name": "streamreader",
"parameter": {"column": [],
"sliceRecordCount": ""
}
},
"writer": {
"name": "streamwriter",
"parameter": {"encoding": "","print": true}
}
}
],
"setting": {
"speed": {"channel": ""}
}
}
}依据模板编写 json 文件
[root@MySQL-1 ~]# cat <<END > test.json
{
"job": {
"content": [
{
"reader": {
"name": "streamreader",
"parameter": {"column": [ # 同步的列名 (* 示意所有)
{
"type":"string",
"value":"Hello."
},
{
"type":"string",
"value":"河北彭于晏"
},
],
"sliceRecordCount": "3" # 打印数量
}
},
"writer": {
"name": "streamwriter",
"parameter": {
"encoding": "utf-8", # 编码
"print": true
}
}
}
],
"setting": {
"speed": {"channel": "2" # 并发 (即 sliceRecordCount * channel = 后果)
}
}
}
}输入：（要是复制我下面的话，须要把 # 带的内容去掉）
3. 装置 MySQL 数据库
别离在两台主机上安装：
[root@MySQL-1 ~]# yum -y install mariadb mariadb-server mariadb-libs mariadb-devel
[root@MySQL-1 ~]# systemctl start mariadb # 装置 MariaDB 数据库
[root@MySQL-1 ~]# mysql_secure_install # 初始化
NOTE: RUNNING ALL PARTS OF THIS SCRIPT IS RECOMMENDED FOR ALL MariaDB
SERVERS IN PRODUCTION USE! PLEASE READ EACH STEP CAREFULLY!
Enter current password for root (enter for none): # 间接回车
OK, successfully used password, moving on...
Set root password? [Y/n] y # 配置 root 明码
New password: 123123
Re-enter new password: 123123
Password updated successfully!
Reloading privilege tables..
... Success!
Remove anonymous users? [Y/n] y # 移除匿名用户
... skipping.
Disallow root login remotely? [Y/n] n # 容许 root 近程登录
... skipping.
Remove test database and access to it? [Y/n] y # 移除测试数据库
... skipping.
Reload privilege tables now? [Y/n] y # 从新加载表
... Success!1）筹备同步数据（要同步的两台主机都要有这个表）
MariaDB [(none)]> create database `course-study`;
Query OK, 1 row affected (0.00 sec)
MariaDB [(none)]> create table `course-study`.t_member(ID int,Name varchar(20),Email varchar(30));
Query OK, 0 rows affected (0.00 sec)因为是应用 DataX 程序进行同步的，所以须要在单方的数据库上凋谢权限：
grant all privileges on *.* to root@'%' identified by '123123';
flush privileges;2）创立存储过程：
DELIMITER $$
CREATE PROCEDURE test()
BEGIN
declare A int default 1;
while (A < 3000000)do
insert into `course-study`.t_member values(A,concat("LiSa",A),concat("LiSa",A,"@163.com"));
set A = A + 1;
END while;
END $$
DELIMITER ;3）调用存储过程（在数据源配置，验证同步应用）：
call test();4. 通过 DataX 实 MySQL 数据同步
1）生成 MySQL 到 MySQL 同步的模板：
[root@MySQL-1 ~]# python /usr/local/datax/bin/datax.py -r mysqlreader -w mysqlwriter
{
"job": {
"content": [
{
"reader": {
"name": "mysqlreader", # 读取端
"parameter": {"column": [], # 须要同步的列 (* 示意所有的列)
"connection": [
{"jdbcUrl": [], # 连贯信息
"table": [] # 连贯表}
],
"password": "", # 连贯用户"username":"", # 连贯明码
"where": "" # 形容筛选条件
}
},
"writer": {
"name": "mysqlwriter", # 写入端
"parameter": {"column": [], # 须要同步的列
"connection": [
{"jdbcUrl": "", # 连贯信息"table": [] # 连贯表
}
],
"password": "", # 连贯明码"preSql": [], # 同步前. 要做的事"session": [],"username":"", # 连贯用户
"writeMode": "" # 操作类型
}
}
}
],
"setting": {
"speed": {"channel": "" # 指定并发数}
}
}
}2）编写 json 文件：
[root@MySQL-1 ~]# vim install.json
{
"job": {
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"username": "root",
"password": "123123",
"column": ["*"],
"splitPk": "ID",
"connection": [
{
"jdbcUrl": ["jdbc:mysql://192.168.1.1:3306/course-study?useUnicode=true&characterEncoding=utf8"],
"table": ["t_member"]
}
]
}
},
"writer": {
"name": "mysqlwriter",
"parameter": {"column": ["*"],
"connection": [
{
"jdbcUrl": "jdbc:mysql://192.168.1.2:3306/course-study?useUnicode=true&characterEncoding=utf8",
"table": ["t_member"]
}
],
"password": "123123",
"preSql": ["truncate t_member"],
"session": ["set session sql_mode='ANSI'"],"username":"root","writeMode":"insert"
}
}
}
],
"setting": {
"speed": {"channel": "5"}
}
}
}3）验证
[root@MySQL-1 ~]# python /usr/local/datax/bin/datax.py install.json输入：
2021-12-15 16:45:15.120 [job-0] INFO JobContainer - PerfTrace not enable!
2021-12-15 16:45:15.120 [job-0] INFO StandAloneJobContainerCommunicator - Total 2999999 records, 107666651 bytes | Speed 2.57MB/s, 74999 records/s | Error 0 records, 0 bytes | All Task WaitWriterTime 82.173s | All Task WaitReaderTime 75.722s | Percentage 100.00%
2021-12-15 16:45:15.124 [job-0] INFO JobContainer -
工作启动时刻 : 2021-12-15 16:44:32
工作完结时刻 : 2021-12-15 16:45:15
工作总计耗时 : 42s
工作均匀流量 : 2.57MB/s
记录写入速度 : 74999rec/s
读出记录总数 : 2999999
读写失败总数 : 0你们能够在目标数据库进行查看，是否同步实现。
- 下面的形式相当于是齐全同步，然而当数据量较大时，同步的时候被中断，是件很苦楚的事件；
- 所以在有些状况下，增量同步还是蛮重要的。
5. 应用 DataX 进行增量同步
应用 DataX 进行全量同步和增量同步的惟一区别就是：增量同步须要应用 where 进行条件筛选。
- 即，同步筛选后的 SQL。
1）编写 json 文件：
[root@MySQL-1 ~]# vim where.json
{
"job": {
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"username": "root",
"password": "123123",
"column": ["*"],
"splitPk": "ID",
"where": "ID <= 1888",
"connection": [
{
"jdbcUrl": ["jdbc:mysql://192.168.1.1:3306/course-study?useUnicode=true&characterEncoding=utf8"],
"table": ["t_member"]
}
]
}
},
"writer": {
"name": "mysqlwriter",
"parameter": {"column": ["*"],
"connection": [
{
"jdbcUrl": "jdbc:mysql://192.168.1.2:3306/course-study?useUnicode=true&characterEncoding=utf8",
"table": ["t_member"]
}
],
"password": "123123",
"preSql": ["truncate t_member"],
"session": ["set session sql_mode='ANSI'"],"username":"root","writeMode":"insert"
}
}
}
],
"setting": {
"speed": {"channel": "5"}
}
}
}- 须要留神的局部就是：
where（条件筛选）和preSql（同步前，要做的事）参数。
2）验证：
[root@MySQL-1 ~]# python /usr/local/data/bin/data.py where.json输入：
2021-12-16 17:34:38.534 [job-0] INFO JobContainer - PerfTrace not enable!
2021-12-16 17:34:38.534 [job-0] INFO StandAloneJobContainerCommunicator - Total 1888 records, 49543 bytes | Speed 1.61KB/s, 62 records/s | Error 0 records, 0 bytes | All Task WaitWriterTime 0.002s | All Task WaitReaderTime 100.570s | Percentage 100.00%
2021-12-16 17:34:38.537 [job-0] INFO JobContainer -
工作启动时刻 : 2021-12-16 17:34:06
工作完结时刻 : 2021-12-16 17:34:38
工作总计耗时 : 32s
工作均匀流量 : 1.61KB/s
记录写入速度 : 62rec/s
读出记录总数 : 1888
读写失败总数 : 0指标数据库上查看：
3）基于下面数据，再次进行增量同步：
- 次要是
where配置："where": "ID > 1888 AND ID <= 2888"（通过条件筛选来进行增量同步） - 同时须要将我下面的
preSql删除 (因为我下面做的操作是 truncate 表)
近期热文举荐：
1.1,000+ 道 Java 面试题及答案整顿(2022 最新版)
2. 劲爆！Java 协程要来了。。。
3.Spring Boot 2.x 教程，太全了！
4. 别再写满屏的爆爆爆炸类了，试试装璜器模式，这才是优雅的形式！！
5.《Java 开发手册（嵩山版）》最新公布，速速下载！
感觉不错，别忘了顺手点赞 + 转发哦！