datax 3.0配合crontab实现数据定时增量同步

使用datax 实现数据增量同步踩坑记录

前提概要

由于项目上需要将a服务器数据同步至b服务器，一开始使用mysql主从复制，但是由于主从同步无法触发位于b服务器的触发器，只能放弃此方案。后来找到了datax可以实现数据同步同时也可以触发触发器，决定使用此方案。

datax准备

• 安装datax，python，jdk  datax下载及安装地址
• 根据自身需求选择合适的writer和reader，我这里选择的是mysqlwriter和mysqlreader
• 下面是我使用的json文件，有两点需要注意 
  
  1. 我在 where 使用了sql 语句  create_time > FROM_UNIXTIME(${create_time}) and create_time < FROM_UNIXTIME(${end_time}) ，其中FROM_UNIXTIME()是mysql时间戳转换为时间格式的函数，${name}是datax提供的占位符后面会使用到
  2. reader中连接字符串添加了 useUnicode=true&characterEncoding=utf8 ，因为没有加这个导入到目标数据库中文乱码了，虽然我两边的数据库都是utf8mb4格式的

   {
   "job": {
     "setting": {
       "speed": {
         "channel": 3
       },
       "errorLimit": {
         "record": 0,
         "percentage": 0.02
       }
     },
     "content": [
       {
         "reader": {
           "name": "mysqlreader",
           "parameter": {
             "username": "root",
             "password": "root",
             "where": "create_time > FROM_UNIXTIME(${create_time}) and create_time < FROM_UNIXTIME(${end_time})",
             "column": [
               "clue_atta_id",
               "url",
               "create_time",
               "atta_type",
               "clue_id",
               "name",
               "attachment_id",
               "attr_sequence"
             ],
             "connection": [
               {
                 "table": [
                   "bus_clue_atta"
                 ],
                 "jdbcUrl": [
                   "jdbc:mysql://x.x.x.x:3306/dbname"
                 ]
               }
             ]
           }
         },
         "writer": {
           "name": "mysqlwriter",
           "parameter": {
             "writeMode": "insert",
             "username": "root",
             "password": "root",
             "column": [
               "clue_atta_id",
               "url",
               "create_time",
               "atta_type",
               "clue_id",
               "name",
               "attachment_id",
               "attr_sequence"
             ],
             "session": [
               "set session sql_mode='ANSI'"
             ],
             "connection": [
               {
                 "jdbcUrl": "jdbc:mysql://x.x.x.x:3306/dbname?useUnicode=true&characterEncoding=utf8",
                 "table": [
                   "bus_clue_atta"
                 ]
               }
             ]
           }
         }
       }
     ]
   }
 }   
  
   

shell脚本准备

• 因为我有多张表，编写一个脚本 
  
  • 需要添加source /etc/profile ，因为在cron的系统环境和shell的环境不一样，会导致 java commond not found错误  参考地址
  • $(date +%s) 为获取系统当前时间戳 ， $(($end_time - 60)) 为算术表达式计算60前的时间戳
  • "-Dcreate_time=$create_time -Dend_time=$end_time"这里就是datax使用占位符的作用，可以将外部自定义参数传入
  • >>/home/gzjp/datax_log/bus_clue_atta_log.date +%Y%m%d 2>&1 我这里把日期都放入每天的日志文件以免单一文件过大
  • 最后由于我的同步的表格较少我直接使用  & 进行后台操作以免发生阻塞

  #!/bin/bash
source /etc/profile
# 截至时间设置为当前时间戳
end_time=$(date +%s)
# 开始时间设置为60s前时间戳
create_time=$(($end_time - 60))
/home/gzjp/datax/bin/datax.py /home/gzjp/jobs/bus_clue_atta_job.json -p "-Dcreate_time=$create_time -Dend_time=$end_time" >>/home/gzjp/datax_log/bus_clue_atta_log.`date +%Y%m%d`  2>&1 &
/home/gzjp/datax/bin/datax.py /home/gzjp/jobs/bus_clue_job.json -p "-Dcreate_time=$create_time -Dend_time=$end_time" >>/home/gzjp/datax_log/bus_clue_log.`date +%Y%m%d`  2>&1 &
/home/gzjp/datax/bin/datax.py /home/gzjp/jobs/bus_attachment.json -p "-Dcreate_time=$create_time -Dend_time=$end_time" >>/home/gzjp/datax_log/bus_attachment_log.`date +%Y%m%d`  2>&1 &
   
  
   

crontab 定时任务准备

  $ crontab -e

*/1 * * * * /home/gzjp/jm_db_sync.sh >/dev/null 2>&1   
  
   

• 我是定时每分钟跑一次脚本， 注意一定要处理输入文件，因为cron会见执行情况通过mail发给用户，时间长系统会被塞爆

有意义的参考内容

https://helpcdn.aliyun.com/document_detail/62149.html