需求：把线下210数据同步到本地中台数据库中
一、datax 简介
datax 是阿里云 DataWorks 数据集成 的开源版本，主要就是用于实现数据间的离线同步。 datax 致力于实现包括关系型数据库（MysqL、Oracle 等）、HDFS、Hive、ODPS、HBase、FTP 等 各种异构数据源（即不同的数据库） 间稳定高效的数据同步功能。
当需要接入一个新的数据源时，只需要将此数据源对接到 datax，便能跟已有的数据源作为无缝数据同步。
1.datax3.0 框架设计
datax 采用 Framework + Plugin 架构，将数据源读取和写入抽象称为 Reader/Writer 插件，纳入到整个同步框架中。
角色 作用
| 2022-04-24 17:21:04.845 [job-0] INFO JobContainer - 任务启动时刻 : 2022-04-24 17:20:54 任务结束时刻 : 2022-04-24 17:21:04 任务总计耗时 : 10s 任务平均流量 : 253.91KB/s 记录写入速度 : 10000rec/s 读出记录总数 : 100000 读写失败总数 : 0 |
|
2.datax3.0 核心架构
datax 完成单个数据同步的作业，我们称为 Job，datax 接收到一个 Job 后，将启动一个进程来完成整个作业同步过程。datax Job 模块是单个作业的中枢管理节点，承担了数据清理、子任务切分、TaskGroup 管理等功能。
接着 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 地址 | 软件包 |
| 线上210 | CentOS 7.9 | 182.169.212.210 | |
| 线下192 | CentOS 7.9 | 192.168.88.192 | jdk-8u181-linux-x64.tar.gz datax.tar.gz |
安装 JDK：下载地址（需要创建 Oracle 账号）
下载jdk解压到/usr/local/java/
#解压jdk到指定目录
tar -zxvf jdk-8u161-linux-x64.tar.gz -C /usr/local/java/
配置jdk环境变量
vi /etc/profile
export JAVA_HOME=/usr/local/java/jdk1.8.0_161
export PATH=$JAVA_HOME/bin:$PATH
export CLAsspATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export CANAL_SERVER_HOME=/usr/local/canal/deployer
export PATH=$CANAL_SERVER_HOME/bin:$PATH
[root@192 ~]# source /etc/profile
[root@192 ~]# java -version
java version "1.8.0_161"
Java(TM) SE Runtime Environment (build 1.8.0_161-b12)
Java HotSpot(TM) 64-Bit Server VM (build 25.161-b12,mixed mode)
[root@192 ~]# javac #能显示java信息表示安装jdk按装成功因为 CentOS 7 上自带 Python 2.7 的软件包，所以不需要进行安装
1.Linux 上安装 datax 软件
[root@192 ~]# wget http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz
[root@192 ~]# tar zxf datax.tar.gz -C /usr/local/
[root@192 ~]# rm -rf /usr/local/datax/plugin/*/._* # 需要删除隐藏文件 (重要)
验证：
[root@192 ~]# cd /usr/local/datax/bin
[root@192 ~]# python datax.py ../job/job.json # 用来验证是否安装成功
输出：
2022-04-24 17:21:04.845 [job-0] INFO JobContainer -
任务启动时刻 : 2022-04-24 17:20:54
任务结束时刻 : 2022-04-24 17:21:04
任务总计耗时 : 10s
任务平均流量 : 253.91KB/s
记录写入速度 : 10000rec/s
读出记录总数 : 100000
读写失败总数 : 0
能显示出来表示验证成功
2.通过 datax 实 MysqL 数据同步
#输出MysqL配置模版
[root@192 bin]# python /usr/local/datax/bin/datax.py -r MysqLreader -w MysqLwriter > /usr/local/datax/job/MysqL2MysqL.json
#根据模板编写 MysqL2MysqL.json文件
{
"job": {
"content": [
{
"reader": {
"name": "MysqLreader","parameter": {
"column": ["id","name"],#"*"表示所有字段
"connection": [
{
"jdbcUrl": ["jdbc:MysqL://x.x.x.210:3306/mytest"],"table": ["user"]
}
],"password": "root","username": "root"
}
},"writer": {
"name": "MysqLwriter","connection": [
{
"jdbcUrl": "jdbc:MysqL://192.168.88.192:3306/mytest","username": "root","writeMode": "insert"
}
}
}
],"setting": {
"speed": {
"channel": "6"
}
}
}
}
验证：
[root@192 job]# python /usr/local/datax/bin/datax.py MysqL2MysqL.json
2022-04-24 17:39:03.445 [job-0] INFO JobContainer -
任务启动时刻 : 2022-04-24 17:38:49
任务结束时刻 : 2022-04-24 17:39:03
任务总计耗时 : 14s
任务平均流量 : 0B/s
记录写入速度 : 0rec/s
读出记录总数 : 3
读写失败总数 : 0
三、datax-WEB 安装部署
1.IDEA编译打包
下载链接：https://pan.baidu.com/s/13a8nIpz6FL8y4fdE94trjQ
提取码：data
备注:官方提供的版本tar版本包
https://pan.baidu.com/s/13yoqhGpD00I82K4lOYtQhg 提取码：cpsk
2.解压安装包
在选定的安装目录，解压安装包
[root@192 ~]# tar -zxvf datax-web-2.1.2.tar.gz -C /usr/local/dataxweb3.登录msyql建库
为接下来一键安装部署准备，这里我建的库是dataxweb(自己定义就好，前后保持一致)
MysqL> create database dataxweb;
4.执行一键安装脚本
进入解压后的目录，找到bin目录下面的install.sh文件，如果选择交互式的安装，则直接执行
[root@192 dataxweb]# cd bin/
[root@192 bin]# pwd
/usr/local/dataxweb/bin
[root@192 bin]# ./install.sh 然后按照提示操作即可。包含了数据库初始化，如果你的服务上安装有MySQL命令，在执行安装脚本的过程中则会出现以下提醒：
Scan out MysqL command,so begin to initalize the database
Do you want to initalize database with sql: [{INSTALL_PATH}/bin/db/datax-web.sql]? (Y/N)y
Please input the db host(default: 127.0.0.1):
Please input the db port(default: 3306):
Please input the db username(default: root):
Please input the db password(default: ): root
Please input the db name(default: dataxweb)按照提示输入数据库地址，端口号，用户名，密码以及数据库名称，大部分情况下即可快速完成初始化。
如果服务上并没有安装MySQL命令，则可以取用目录下/bin/db/datax-web.sql脚本去手动执行，完成后修改相关配置文件
vi modules/datax-admin/conf/bootstrap.properties
#Database
DB_HOST=127.0.0.1
DB_PORT=3306
DB_USERNAME=root
DB_PASSWORD=root
DB_DATABASE=dataxweb按照具体情况配置对应的值即可。
在交互模式下，对各个模块的package压缩包的解压以及configure配置脚本的调用，都会请求用户确认,可根据提示查看是否安装成功，如果没有安装成功，可以重复尝试； 如果不想使用交互模式，跳过确认过程，则执行以下命令安装
./bin/install.sh --force
4.其他配置
4.1邮件服务
在项目目录：modules/datax-admin/bin/env.properties 配置邮件服务(可跳过)
MAIL_USERNAME=""
MAIL_PASSWORD=""此文件中包括一些默认配置参数，例如：server.port，具体请查看文件。
4.2指定PYTHON_PATH的路径
vim modules/datax-executor/bin/env.properties
### 执行datax的python脚本地址
PYTHON_PATH=/usr/local/datax/bin/datax.py
### 保持和datax-admin服务的端口一致；默认是9527，如果没改datax-admin的端口，可以忽略
datax_ADMIN_PORT=此文件中包括一些默认配置参数，例如：executor.port,json.path,data.path等，具体请查看文件。
5.启动服务
5.1一键启动所有服务
[root@192 dataxweb]# cd /usr/local/dataxweb/
[root@192 dataxweb]# ./bin/start-all.sh 中途可能发生部分模块启动失败或者卡住，可以退出重复执行，如果需要改变某一模块服务端口号，则：
vi ./modules/{module_name}/bin/env.properties
找到SERVER_PORT配置项，改变它的值即可。 当然也可以单一地启动某一模块服务：
./bin/start.sh -m {module_name}
module_name可以为datax-admin或datax-executor
5.2一键停止所有服务
[root@192 dataxweb]# cd /usr/local/dataxweb/
[root@192 dataxweb]# ./bin/stop-all.sh当然也可以单一地停止某一模块服务：
./bin/stop.sh -m {module_name}
5.3查看服务（注意！注意！）
在Linux环境下使用JPS命令，查看是否出现dataxAdminApplication和dataxExecutorApplication进程，如果存在这表示项目运行成功
如果项目启动失败，请检查启动日志：
modules/datax-admin/bin/console.out
或者
modules/datax-executor/bin/console.out四、datax-WEB 运行
1.前端界面
部署完成后，在浏览器中输入 http://ip:port/index.html 就可以访问对应的主界面（ip为datax-admin部署所在服务器ip,port为为datax-admin 指定的运行端口9527）
输入用户名 admin 密码 123456 就可以直接访问系统
2.datax-web API
datax-web部署成功后，可以了解datax-web API相关内容，网址: http://ip:port/doc.html
五、datax-WEB 运行日志
部署完成之后，在modules/对应的项目/data/applogs下(用户也可以自己指定日志，修改application.yml 中的logpath地址即可)，用户可以根据此日志跟踪项目实际启动情况
如果执行器启动比admin快，执行器会连接失败，日志报"拒绝连接"的错误，一般是先启动admin,再启动executor,30秒之后会重连，如果成功请忽略这个异常。
六、datax-WEB 实操
1.查看执行器
查看web界面是否有注册成功的执行器，另外执行器可以根据需要改名称。
3.任务模板构建
3.2路由策略：当执行器集群部署时，提供丰富的路由策略，包括:
FirsT（第一个）：固定选择第一个机器；
LAST（最后一个）：固定选择最后一个机器；
ROUND（轮询）：依次分配任务；
RANDOM（随机）：随机选择在线的机器；
CONSISTENT_HASH（一致性HASH）：每个任务按照Hash算法固定选择某一台机器，且所有任务均匀散列在不同机器上。
LEAST_FREQUENTLY_USED（最不经常使用）：使用频率最低的机器优先被选举；
LEAST_RECENTLY_USED（最近最久未使用）：最久为使用的机器优先被选举；
FAIlovER（故障转移）：按照顺序依次进行心跳检测，第一个心跳检测成功的机器选定为目标执行器并发起调度；
BUSYOVER（忙碌转移）：按照顺序依次进行空闲检测，第一个空闲检测成功的机器选定为目标执行器并发起调度；
阻塞处理策略：调度过于密集执行器来不及处理时的处理策略；
- 单机串行：调度请求进入单机执行器后，调度请求进入FIFO队列并以串行方式运行；
- 丢弃后续调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败；
- 覆盖之前调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，将会终止运行中的调度任务并清空队列，然后运行本地调度任务；
增量增新建议将阻塞策略设置为丢弃后续调度或者单机串行
设置单机串行时应该注意合理设置重试次数(失败重试的次数*每次执行时间<任务的调度周期)，重试的次数如果设置的过多会导致数据重复，例如任务30秒执行一次，每次执行时间需要20秒，设置重试三次，如果任务失败了，第一个重试的时间段为1577755680-1577756680，重试任务没结束，新任务又开启，那新任务的时间段会是1577755680-1577758680
3.3任务类型
先选择datax任务，后续配置完详细任务后可以更改3.4其它
可以根据需求填写
3.4其它
4.数据源配置
根据不同数据源，配置参数。
5.任务构建
构建reader
{
"job": {
"content": [
{
"reader": {
"name": "MysqLreader","parameter": {
"username": "root","column": [
"*"
],"where": " save_time >= FROM_UNIXTIME(${lastTime}) and save_time < FROM_UNIXTIME(${currentTime})","splitPk": "id","connection": [
{
"table": [
"uc_op_amazon_api_store_download"
],"jdbcUrl": [
"jdbc:MysqL://x.x.x.210:3306/test_system"
]
}
]
}
},"writer": {
"name": "MysqLwriter","parameter": {
"writeMode": "insert","connection": [
{
"jdbcUrl": "jdbc:MysqL://192.168.88.192:3306/mytest?useUnicode=true&characterEncoding=utf8","table": [
"uc_op_amazon_api_store_download"
]
}
]
}
}
}
],"setting": {
"speed": {
"channel": 6
}
}
}
}-DstartId='%s' -DendId='%s'
# 表名
uc_op_business_reports
#主键
id
#id自增配置条件
"where": " id>= ${startId} and id<= ${endId}",
入库前 清表
"presql": [
"truncate table uc_op_dynnamic_ad_api_day"
],
问题1：
配置远程数据源过程中出现异常，提示远程210不能连接
Code:[MysqLErrCode-02],Description:[数据库服务的IP地址或者Port错误，请检查填写的IP地址和Port或者联系DBA确认IP地址和Port是否正确。如果是同步中心用户请联系DBA确认idb上录入的IP和PORT信息和数据库的当前实际信息是一致的]. - 具体错误信息为：com.MysqL.jdbc.exceptions.jdbc4.CommunicationsException: Communications link failure The last packet successfully received from the server was 174 milliseconds ago. The last packet sent successfully to the server was 174 milliseconds ago.
把./datax-admin/lib/mysql-connector-java-5.1.47.jar替换成mysql-connector-java-5.1.49.jar好了
问题2：
2022-04-28 16:00:08.872 [New I/O server worker #1-3] ERROR c.a.otter.canal.server.netty.handler.SessionHandler - something goes wrong with channel:[id: 0x31df79de,/211.103.136.38:44900 => /x.x.x.210:11111],exception=java.io.IOException: Connection reset by peer
at sun.nio.ch.FiledispatcherImpl.read0(Native Method)
at sun.nio.ch.socketdispatcher.read(Socketdispatcher.java:39)
at sun.nio.ch.IoUtil.readIntoNativeBuffer(IoUtil.java:223)
at sun.nio.ch.IoUtil.read(IoUtil.java:192)
at sun.nio.ch.socketChannelImpl.read(SocketChannelImpl.java:380)
at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:322)
at org.jboss.netty.channel.socket.nio.NioWorker.processSelectedKeys(NioWorker.java:281)
at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:201)
at org.jboss.netty.util.internal.IoWorkerRunnable.run(IoWorkerRunnable.java:46)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。
