DataX
- github地址：DataX/introduction.md at master · alibaba/DataX · GitHub
环境准备
-
Linux环境系统
-
JDK（1.8及其以上版本，推荐1.8）
-
Python（2或者3都可以）
-
Apache Maven 3.x（源码编译安装需要）
安装部署
二进制安装
-
下载地址：https://datax-opensource.oss-cn-hangzhou.aliyuncs.com/202308/datax.tar.gz
-
上传之Linux服务器指定位置
-
解压该文件
tar -zxvf datax.tar.gz -
自检文件
python /home/opt/datax/datax/bin/datax.py /home/opt/datax/datax/job/job.json
python3支持
-
将bin目录下的三个py文件进行修改；
-
print xxx 替换为 print(xxx)
-
Exccption, e替换为 Exception as e
-
自检文件
python3 /home/opt/datax/datax/bin/datax.py /home/opt/datax/datax/job/job.json
模版文件生成
python3 /home/opt/datax/datax/bin/datax.py -r mysqlreader -w hdfswriter > ~/home/opt/datax/datax/job/mysql2hdfs.json
自定义同步脚本
{"job": {"setting": {"speed": {"channel": 1}},"content": [{"reader": {"name": "streamreader","parameter": {"column": [{"value": "hello world","type": "string"},{"value": "datax","type": "string"},{"value": 12312312,"type": "long"}],"sliceRecordCount": 1 // 读取数据列中的重复次数}},"writer": {"name": "streamwriter","parameter": {"print": true}}}]}
}python3 /home/opt/datax/datax/bin/datax.py /home/opt/datax/datax/job/jobs/job1.json
动态传参
-
将mysql数据同步到HDFS，多次同步的时候只是表的名字和字段不同。
-
将mysql的数据增量同步到HDFS或者Hive中的时候，需要指定每一次同步的时间。
-
所谓的动态传参，就是在json的同步方案中，使用类似变量的方式来定义一些可以改变的参数，在执行同步方案的时候，可以指定这些参数具体的值。
动态传参的案例
{
“job”: {
“setting”: {
“speed”: {
“channel”: 1
}
},
“content”: [
{
“reader”: {
“name”: “streamreader”,
“parameter”: {
“column”: [
{“value”: “hello world”, “type”: “string”},
{“value”: “datax”, “type”: “string”},
{“value”: 12312312, “type”: “int”}
],
“sliceRecordCount”: $TIMES // 读取数据列中的重复次数
}
},
“writer”: {
“name”: “streamwriter”,
“parameter”: {
“print”: true
}
}
}
]
}
}python3 /home/opt/datax/datax/bin/datax.py -p “-DTIMES=3” /home/opt/datax/datax/job/jobs/job1.json
并发设置
-
直接指定Channel数量
-
通过Bps计算channel数量
-
通过tps计算channel数量
直接指定
在同步方案的json文件中，可以设置job.setting.speed.channel来设置channel的数量，这是最直接的方式，在这种配置下，channel的Bps为默认的1MBps，即每秒传输1MB的数据。
Bps
Bps(Byte per second) 是一种非常常见的数据传输速率的表示，在DataX中，可以通过参数设置来限制总job的Bps以及单个channel的Bps，来达到限速和channel数量计算的效果。
channel = Job Bps/channel Bps
-
Job Bps: 对一个job进行整体的限速，可以通过job.setting.speed.byte进行设置
-
channel Bps: 对单个channel的限速，可以通过core.transport.channel.speed.byte进行设置
tps
tps(transcation per second)是一种很常见的数据传输速率的表示，在DataX中，可以通过参数设置来限制总Job的tps以及单个channel的tps，来达到限速和channel数量计算的效果。
channel= job tps/channel tps
优先级
-
如果同时配置了Bps和tps限制，以下的为准。
-
只有在Bps和tps都没有配置的时候，才会以channel数量配置为准。
stream
streamreader
参数说明
参数
描述
column
定义内存数据的所有列。由一个JSON组成，需要滴定仪数据与类型。
sliceRecordCount
每个channel中，数据重复的数量。
数据类型
在定义每一个列的数据的时候，需要指定每一列的数据类型。DataX中支持如下的数据类型：
-
Long
-
Double
-
String
-
Date
-
Boolean
-
Bytes
streamwriter
参数说明
参数
描述
将内存中的数据打印在控制台上。
encoding
输出数据使用的字符集，默认是UTF-8。
mysql
mysqlreader
介绍
MysqlRader插件实现了从mysql读取数据，在底层实现上，MySQLReader通过JDBC远程MySQL数据库，并执行相应的SQL数据，将数据从MySQL数据库中查询出来。
简而言之，MySQLReader通过JDBC连接器连接到远程的MySQL数据库，并根据用户配置的信息生成查询的SQL语句，然后发送到远程MySQL数据库，并将该SQL执行返回结果使用DataX自定义的数据离线拼装为抽象的数据集，并传递给下游writer处理。
对于用户配置Table、column、where的信息，MySQLReader将其拼接成SQL语句发送到MySQL数据库；对于用户配置querySQL信息，MySQLReader直接将其发送到MySQL数据库；
参数说明
参数
描述
必选
默认值
jdbcUrl
描述的是对端数据库的JDBC连接信息，使用JSON的数组描述，并支持一个库填写多个连接地址。之所以使用JSON数组描述连接信息，是因为阿里集团内部支持多个IP探测，如果配置了多个，MySQLReader可以一次探测IP的可能性，直到选择一个合法的IP。如果全部连接失败，MySQLReader报错。注意，jdbcUrl必须包含在connection配置单元中。对于阿里集团外部使用情况，JSON数组填写一个JDBC连接即可。
是
无
username
连接到数据库的用户名
是
无
password
连接到数据库的密码
是
无
table
所选取的需要同步的表。使用JSON的数组描述，因此支持多张表同时抽取。当配置为多张表时，用户自己需要保证多张表是同一个schema结构，MySQLReader不予检查表是否同一逻辑表。注意，table必须包含在connection的配置单元中。
是
无
column
所配置的表中需要同步的列名集合，使用JSON数组描述字段信息。用户使用*代表默认使用所有列配置。
支持列裁剪，即列可以挑选部分列进行导出。
支持列换序，即列可以不按照schema信息进行导出。
支持常量配置，用户需要按照MySQL语法格式。[“id”, “tabke”, “1”]
类型转换
MySQLReader与MySQL的类型比较
案例
{"job": {"setting": {"speed": {"channel": 3},"errorLimit": {"record": 0,"percentage": 0.02}},"content": [{"reader": {"name": "mysqlreader","parameter": {"username": "root","password": "123456","column": ["id","username","password","age","sex","phone"],"splitPk": "username","connection": [{"table": ["admin"],"jdbcUrl": ["jdbc:mysql://192.168.42.130:3306/springboot-backend"]}]}},"writer": {"name": "streamwriter","parameter": {"print": true}}}]}
}
执行方法
python3 /home/opt/datax/datax/bin/datax.py mysql2stream.json
错误处理
1、在/dataX/plugin/reader/mysqlreader/libs中，把mysql-connector-java-5.1.34.jar包删除
rm -rf mysql-connector-java-5.1.34.jar
2、将MySQL8相关的jar包复制到该目录下
mysqlreader读取模式
mysqlwriter
介绍
mysqlwriter插件实现了写入数据到mysql数据库的目的表的功能。在底层实现上，mysqlwriter通过JDBC连接到远程数据库，并执行相应的inster into…或者replace into…的SQL语句，将数据写入到MySQL的数据库，内部会分批次提交入库，需要数据库本身采用innodb引擎。
MySQLwriter面向ETL开发工程师，使用MySQLwriter从数仓导入数据到MySQL，同时MySQLwriter也可以作为数据迁移工具，为DBA等用户提供服务。
实现原理
MySQLwriter通过DataX框架获取Reader生成的协议数据，根据你配置的writerMode生成
-
Insert： Insert into … (当主键、唯一性索引冲突时会写不进去冲突的行)
-
replace：replace into…（没有遇到主键、唯一性索引冲突时，与insert info行为一致，冲突时会用新行替换原有行的所有字段）
-
update：Insert into… on duplicate key update…(如果 on duplicate key update的子句中要更新的值为原来的值都一样，则不更新，否则就更新)
生成上述的雨具写入数据到MySQL，处于性能考虑，采用PreparedStatement+Batch，并且设置了rewriteBatchedStatement=true，将数据缓冲到线程上下文buffer中，当buffer累计到预定阈值时，才发起写入请求。
注意
目标表所在数据库必须要主表才能写入数据；整合任务至少需要具备insert into/replace into的权限，是否需要其他权限，取决于你任务配置中在presql和postsql中指定的语句。
参数说明
错误处理
1、在/dataX/plugin/reader/mysqlreader/libs中，把mysql-connector-java-5.1.34.jar包删除
rm -rf mysql-connector-java-5.1.34.jar
2、将MySQL8相关的jar包复制到该目录下
案例
{"job": {"setting": {"speed": {"channel": 3}},"content": [{"reader": {"name": "mysqlreader","parameter": {"username": "root","password": "123456","column": ["id","username","password","age","sex","phone"],"splitPk": "username","connection": [{"table": ["admin"],"jdbcUrl": ["jdbc:mysql://192.168.42.130:3306/springboot-backend"]}]}},"writer": {"name": "mysqlwriter","parameter": {"username": "root","password": "123456","column": ["id","username","password","age","sex","phone"],"writeMode": "replace","splitPk": "username","connection": [{"table": ["admin_bak"],"jdbcUrl": "jdbc:mysql://192.168.42.130:3306/springboot-backend"}]}}}]}
}
执行
python3 /home/opt/datax/datax/bin/datax.py mysql2mysql.json
shell脚本
#!/bin/bash# 设置需要同步的哪一天的数据
# 一般情况下，当天产生的数据，会在第二天的凌晨增量导入到数据仓库中
dt=`date -d yesterday +"%Y-%m-%d"`# 提取时间分量
year=${dt:0:4}
month=${dt:5:2}#设置DataX的路径
DATAX_HOME=/home/opt/datax/datax#设置jobs的路径
JOBS_HOME=/home/opt/datax/datax/job/jobs# 增量同步数据
python3 $DATAX_HOME/bin/datax.py -p "-Ddate=$dt" $JOBS_HOME/mysql2mysql.json
DataX | MySQL多表全量同步和多表定时增量同步_datax同步多张表-CSDN博客
crond服务的安装与管理
-
安装crond服务工具
yum install crontabs #安装，注意包名
-
开启crond服务
sudo systemctl start crond #立即启动
crond服务
sudo systemctl enable crond #设置crond服务开机自启
sudo systemctl enable --now crond #设置开机自启同时启动 -
关闭crond服务
sudo systemctl stop crond #立即停止
crond服务
sudo systemctl disable crond #取消crond服务的开机自启
sudo systemctl disable --now crond #取消开机自启同时停止 -
查看crond服务的状态
sudo systemctl status crond #该命令会显示 crond 服务当前的运行状态以及是否已启用开机自启
-
重启crond服务
sudo systemctl restart crond #重启 crond 服务
配置定时任务脚本
-
添加定时器，脚本授权
chmod 777 shell1.sh -
编辑定时器
*/1 * * * * /home/opt/datax/datax/job/shells/shell1.sh > /home/opt/datax/datax/job/shells/log.log -
查看crontab日志
tail -f /var/log/cron 或 tail -f /var/spool/mail/root -
修改shell1.sh脚本
#!/bin/bash . /etc/profile # 设置需要同步的哪一天的数据 # 一般情况下，当天产生的数据，会在第二天的凌晨增量导入到数据仓库中 dt=`date -d yesterday +"%Y-%m-%d"`# 提取时间分量 year=${dt:0:4} month=${dt:5:2}#设置DataX的路径 DATAX_HOME=/home/opt/datax/datax#设置jobs的路径 JOBS_HOME=/home/opt/datax/datax/job/jobs# 增量同步数据 python3 $DATAX_HOME/bin/datax.py -p "-Ddate=$dt" $JOBS_HOME/mysql2mysql.json
DataX-Web
1、datax-web介绍
1.1、datax-web是什么
项目地址： GitHub - WeiYe-Jing/datax-web: DataX集成可视化页面，选择数据源即可一键生成数据同步任务，支持RDBMS、Hive、HBase、ClickHouse、MongoDB等数据源，批量创建RDBMS数据同步任务，集成开源调度系统，支持分布式、增量同步数据、实时查看运行日志、监控执行器资源、KILL运行进程、数据源信息加密等。
datax-web是一个在DataX之上开发的分布式数据同步工具，提供简单易用的操作界面，降低该用户使用DataX的学习成本，缩短任务配置事件，避免配置过程中出错。用户可以通过页面选择数据源，即可创建数据同步任务。支持RDBMS、Hive、HBase、ClickHouse、MongDB等数据源。RDBMS数据源可以批量创建数据同步任务，支持实时查看同步进度及日志，并提供终止功能同步，集成二次开发xxl-job,可根据时间、自增主键增量同步数据。
任务“执行器”支持集群部署，支持执行器多节点路由策略选择，支持超时控制、失败重试、失败警告、任务依赖、执行器CPU、内存、负载的监控等等。
1.2、datax-web架构
2、datax-web部署
2.1、环境要求
2.2、按照
2.2.1、二进制安装
直接将安装包下载，解压安装到指定路径
https://github.com/WeiYe-Jing/datax-web/archive/refs/tags/v-2.1.2.tar.gz
tar -zxvf datax-web-2.1.2.tar.gz
在/bin下执行install.sh文件
sh install.sh
如果不想交互式执行，直接执行
sh install.sh -force
修改数据库配置
/datax-web-2.1.2/modules/datax-admin/conf/bootstrap.properties
配置
安装完成之后，在项目目录： /datax-web-2.1.2/modules/datax-admin/bin/env.properties 配置邮件服务（可跳过）
# mail account
MAIL_USERNAME=""
MAIL_PASSWORD=""
此文件中包括一些默认配置参数，例如：server.port，具体查看文件。
在项目目录下/datax-web-2.1.2/modules/datax-executor/bin/env.properties执行PYTHON_PATH的路径
# environment variables#JAVA_HOME=""SERVICE_LOG_PATH=${BIN}/../logs
SERVICE_CONF_PATH=${BIN}/../conf
DATA_PATH=${BIN}/../data## datax json文件存放位置
JSON_PATH=${BIN}/../json## executor_port
EXECUTOR_PORT=9999## 保持和datax-admin端口一致
DATAX_ADMIN_PORT=## PYTHON脚本执行位置
#PYTHON_PATH=/home/hadoop/install/datax/bin/datax.py
PYTHON_PATH=/home/opt/datax/datax/bin/datax.py## dataxweb 服务端口
SERVER_PORT=9504#PID_FILE_PATH=${BIN}/service.pid#debug 远程调试端口
#REMOTE_DEBUG_SWITCH=true
#REMOTE_DEBUG_PORT=7004
启动
一键启动所有服务
sh /home/opt/datax/datax-web-2.1.2/bin/start-all.sh
启动成功后会显示两个进程
一键停用所有服务
sh /home/opt/datax/datax-web-2.1.2/bin/stop-all.sh
WebUI
部署完成后，在浏览器输入http://ip:9527/index.html就可以访问对应的直接面（IP为datax-admin部署所在的服务器IP，port为datax-admin指定的运行端口）
输入用户名 admin 密码 123456就可以直接访问系统
