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

标签: | 发表时间:2019-08-18 10:22 | 作者:
出处:https://blog.csdn.net

使用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

相关 [datax crontab 数据] 推荐:

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

- -
使用datax 实现数据增量同步踩坑记录. 由于项目上需要将a服务器数据同步至b服务器,一开始使用mysql主从复制,但是由于主从同步无法触发位于b服务器的触发器,只能放弃此方案. 后来找到了datax可以实现数据同步同时也可以触发触发器,决定使用此方案. 安装datax,python,jdk  datax下载及安装地址.

DataX:实现不同数据库数据同步

- - 标点符
目前成熟的数据导入导出工具比较多,但是一般都只能用于数据导入或者导出,并且只能支持一个或者几个特定类型的数据库. 这样带来的一个问题是,如果拥有很多不同类型的数据库/文件系统(Mysql/Oracle/Rac/Hive/Other…),并且经常需要在它们之间导入导出数据,那么我们可能需要开发/维护/学习使用一批这样的工具(jdbcdump/dbloader/multithread/getmerge+sqlloader/mysqldumper…).

基于datax的数据同步平台 - 黄小雪 - 博客园

- -
为此开发了一个数据同步平台,将同步任务的增删改查、执行的历史日志全部放到平台里,然后交给DBA去自己去操作.          市面上也有一些ETL工具,比如kettle,但是为了练手决定重新造轮子.           平台主要用于数据同步、数据处理等等ETL操作.           平台基于阿里的开源同步工具datax3.0开发.

Linux Crontab 定时任务

- - CSDN博客推荐文章
crontab命令的功能是在一定的时间间隔调度一些命令的执行. 1.1 /etc/Crontab文件. 在/etc目录下有一个crontab文件,这里存放有系统运行的一些调度程序. 每个用户可以建立自己的调度crontab(在/var/spool/cron目录下). 以下是我本机上的crontab文件,.

Linux下通过crontab调度脚本时脚本所使用的环境变量问题

- - CSDN博客推荐文章
通过crontab调度脚本时脚本使用的环境变量与直接在shell中运行脚本时使用的环境变量不同(因为crontab并不知道你所使用的shell,只是简单的设置了HOME、LOGNAME、SHELL和有限的PATH等环境变量),因而可能导致“XXX command not find”的错误或者运行结果不一致的情况.

数据仓库

- Ran - Linux@SOHU
翻译:马少兵、曾怀东、朱翊然、林业. 尽管服务器存储、处理能力得到有效的提高,以及服务器价格的降低,让人们能够负担起大量的服务器,但是商业软件应用和监控工具快速的增加,还是使得人们被大量的数据所困扰. 在数据仓库领域中的许多系统管理员、应用开发者,以及初级数据库管理员发现,他们正在处理“海量数据”-不管你准备与否-都会有好多不熟悉的术语,概念或工具.

数据抽取

- - 数据库 - ITeye博客
转自: http://wiki.mbalib.com/wiki/%E6%95%B0%E6%8D%AE%E6%8A%BD%E5%8F%96#.   数据抽取是指从源数据源系统抽取目的数据源系统需要的. 实际应用中,数据源较多采用的是. 数据迁移或数据复制,它将数据源中的表或视图的数据原封不动的从数 据库中抽取出来,并转换成自己的ETL 工具可以识别的格式.

数据库sharding

- - 数据库 - ITeye博客
当团队决定自行实现sharding的时候,DAO层可能是嵌入sharding逻辑的首选位置,因为在这个层面上,每一个DAO的方法都明确地知道需要访问的数据表以及查询参数,借助这些信息可以直接定位到目标shard上,而不必像框架那样需要对SQL进行解析然后再依据配置的规则进行路由. 另一个优势是不会受ORM框架的制约.

数据脱敏

- - IT瘾-bigdata
作者|李呈祥,其中部分内容由十一城补充. 数据脱敏(Data Masking),又称数据漂白、数据去隐私化或数据变形. 百度百科对数据脱敏的定义为:指对某些敏感信息通过脱敏规则进行数据的变形,实现敏感隐私数据 的可靠保护. 这样,就可以在开发、测试和其它非生产环境以及外包环境中安全地使用脱敏后的真实数据集.

数据分析之如何用数据?

- - 互联网分析沙龙
光知道怎么看数据,还是不成,你得熟悉这些数据拿到手上之后怎么去用它,怎么让数据显示出来它本身的威力来. 第一个部分,是看历史数据,发现规律. 以社区中的活动和电商中的促销为例,这些都是常见的活动,活动做得好的话有意想不到的效果. 在做这样的活动,最好是拿到前一个月或者两个月的历史数据. 对电商来说,从这里面要去分析各个品类的销售情况,那个品类销量最大,那个品类销量最小,每月或者每周的平均增长率和符合增长率是多少.