datax数据同步说明

ETL⼯具DataX数据同步，LINUXCRONTAB定时调度DataX 是阿⾥巴巴集团内被⼴泛使⽤的离线数据同步⼯具/平台，实现包括 MySQL、Oracle、SqlServer、Postgre、HDFS、Hive、ADS、HBase、OTS、ODPS 等各种异构数据源之间⾼效的数据同步功能。

例⼦：全量从MYSQL 同步到MYSQL{ "job": { "content":[ { "reader":{ "name":"mysqlreader", "parameter":{ "connection":[ { "querySql":["SELECT * FROM TABEL"], "jdbcUrl":["jdbc:mysql://127.0.0.1:3306/dsjglpt"], } ], "username":"root", "password":"123456", "mandatoryEncoding":"UTF-8" } }, "writer":{ "name":"mysqlwriter", "parameter":{ "column": ["*"], "connection":[ { "jdbcUrl":"jdbc:mysql://127.0.0.1:3306/dsjglpt_tmp", "table":["TABEL"], } ], "username":"root", "password":"123456", "mandatoryEncoding":"UTF-8", "preSql": ["truncate table TABEL"], "writeMode": "insert" } } } ], "setting":{ "errorLimit":{ "record":0 }, "speed":{ "channel":"1" } } }}从 MYSQL 同步到 PostgreSQL ，部分配置如下："name": "postgresqlwriter""jdbcUrl": "jdbc:postgresql://[target_server]:5432/[target_db]",编写 ty_commit_datax_sjyc.sh （并发任务数）if [ $# -lt 1 ] ; thenecho "请输⼊正确的参数:并发任务数 "exit 1;fiprocessNum=$1##基础路径base=/home/admin##新版本的dataxdataxHome=/home/admin/datax/datax3##配置⽂件⽬录confBase=`ls ${base}/sjyc/*.json`##全量json⽬录qlJsonPath="${base}/sjyc"##如果统计表记录数⽂件不存在，需先执⾏统计表记录数脚本#if [ ! -f ${shellPath}/log/count/${syscode}_CountTable.conf ] ;then# echo "表记录数⽂件${syscode}_CountTable.conf不存在，请先执⾏统计记录数脚本!"# exit 1;#fistartTime=`date '+%Y%m%d%H%M%S'`##输出⽇志路径outpath="${base}/log/commit/sjyc/${startTime}"if [ -d ${outpath} ] ;thenrm -rf ${outpath}fimkdir -p ${outpath}touch ${outpath}/commitJson.logtouch ${outpath}/succJson.logtouch ${outpath}/failJson.logmkdir -p ${outpath}/succLogmkdir -p ${outpath}/failLogjsonNum=`ls ${qlJsonPath}/*.json | wc -l`if [ ${processNum} -gt ${jsonNum} ] ;thenecho "并发任务数必须⼩于等于json⽂件数:${jsonNum}"exit 1;fi##以当前进程号命名fifoName="/tmp/$$.fifo"mkfifo ${fifoName}##定义⽂件描述符(fd是⼀个整数，可理解为索引或指针),以读写的⽅式绑定到⽂件描述符3中exec 3<>"${fifoName}"##此时可以删除管道⽂件，保留fd即可rm -rf ${fifoName}##定义进程数量,向管道⽂件中写⼊进程数量个空⾏for ((i=1;i<=${processNum};i++)) doecho >&3done##组装dataX 动态参数##根据配置⽂件ty_createJson.conf中的表匹配json⽬录下的json⽂件，如果⼀个表匹配到多个json⽂件(分区表的情况)，则依次提交for file in `ls ${qlJsonPath}/*.json` ;do##读⼊⼀个空⾏read -u3{jsonName="${file##*/}"fileName="${jsonName%%.*}"echo ${jsonName}echo "${file}" >> ${outpath}/commitJson.logpython ${dataxHome}/bin/datax.py ${file} 2>&1 >> ${outpath}/${fileName}.logresult=$?if [ ${result} -eq 0 ] ;thenecho "${fileName}" >> ${outpath}/succJson.logmv -f ${outpath}/${fileName}.log ${outpath}/succLogelseecho "${fileName}" >> ${outpath}/failJson.logmv -f ${outpath}/${fileName}.log ${outpath}/failLog##记录失败的表配置#echo "${line}" >> ${confBase}/ty_createJson_bsj.conffi#sleep 1##最后向管道⽂件中写⼊⼀个空⾏echo >&3}&donedonewaitecho "开始时间:${startTime}" >> ${outpath}/commitJson.logecho "开始时间：${startTime}"endTime=`date '+%Y%m%d%H%M%S'`echo "结束时间:${endTime}" >> ${outpath}/commitJson.logecho "结束时间:${endTime}"#关闭⽂件描述符的读exec 3<&-#关闭⽂件描述符的写exec 3>&-exit 0编写CRONTAB 脚本(每天6点半执⾏每次执⾏10个JSON⽂件)crontab-e30 6 * * * /home/admin/ty_commit_datax_sjyc.sh 10------------------------------------------------------------------要实现增量更新，⾸先要 PostgresqlReader 从⽬标数据库读取最⼤⽇期，并⽤ TextFileWriter 写⼊到⼀个 csv ⽂件，这⼀步我的配置如下所⽰：{"job": {"content": [{"reader": {"name": "postgresqlreader","parameter": {"connection": [{"jdbcUrl": ["jdbc:postgresql://[target_server]:5432/[target_db]"],"querySql": ["SELECT max(data_time) FROM public.minute_data"]}],"password": "...","username": "..."}},"writer": {"name": "txtfilewriter","parameter": {"dateFormat": "yyyy-MM-dd HH:mm:ss","fileName": "minute_data_max_time_result","fileFormat": "csv","path": "/scripts/","writeMode": "truncate"}}}],"setting": { }}}最后编写增量同步的 shell ⽂件，内容如下：#!/bin/bash### every exit != 0 fails the scriptset -e# 获取⽬标数据库最⼤数据时间，并写⼊⼀个 csv ⽂件docker run --interactive --tty --rm --network compose --volume $(pwd):/scripts \ beginor/datax:3.0 \/scripts/minute_data_max_time.jsonif [ $? -ne 0 ]; thenecho "minute_data_sync.sh error, can not get max_time from target db!"exit 1fi# 找到 DataX 写⼊的⽂本⽂件，并将内容读取到⼀个变量中RESULT_FILE=`ls minute_data_max_time_result_*`MAX_TIME=`cat $RESULT_FILE`# 如果最⼤时间不为 null 的话，修改全部同步的配置，进⾏增量更新；if [ "$MAX_TIME" != "null" ]; then# 设置增量更新过滤条件WHERE="DataTime > '$MAX_TIME'"sed "s/1=1/$WHERE/g" minute_data.json > minute_data_tmp.json# 将第 45 ⾏的 truncate 语句删除；sed '45d' minute_data_tmp.json > minute_data_inc.json# 增量更新docker run --interactive --tty --rm --network compose --volume $(pwd):/scripts \ beginor/datax:3.0 \/scripts/minute_data_inc.json# 删除临时⽂件rm ./minute_data_tmp.json ./minute_data_inc.jsonelse# 全部更新docker run --interactive --tty --rm --network compose --volume $(pwd):/scripts \ beginor/datax:3.0 \/scripts/minute_data.jsonfi。

DataX简单使用一、安装二、配置1. reader配置```json"reader":"name": "mysqlreader","parameter":"username": "root","connection":"querySql": ["SELECT * FROM table_name"] }],"column": ["id", "name", "age"],"splitPk": "","where": ""}```2. writer配置```json"writer":"name": "mysqlwriter","parameter":"username": "root","connection":"jdbcUrl": ["jdbc:mysql://localhost:3306/dest_db"],"table": ["dest_table_name"]}]}```三、运行配置好DataX后，就可以使用命令行来运行了。

```shellpython bin/datax.py job/job.json```其中，"job.json"是DataX的配置文件，根据自己的实际配置进行修改。

运行命令后，DataX将会根据配置文件进行数据同步。

dataX研究说明目录1.dataX概述 (1)2.部署dataX (1)2.1 安装dataX前检查 (1)2.2mysql数据库安装 (2)2.3oracle数据库安装 (3)2.4dataX安装 (8)3.oracle数据库导入到mysql数据库用例说明 (9)附录1:数据库操作语句 (16)1.dataX概述DataX是阿里巴巴集团内被广泛使用的离线数据同步工具/平台，实现包括OceanBase、MySQL、Oracle、HDFS、Hive、HBase、OTS、ODPS等各种异构数据源之间高效的数据同步功能。

2.部署dataX2.1 安装dataX前检查在安装DataX前需要检查安装的服务器是否满足以下内容：①处理器为―X86_64‖。

②Python版本>=2.4。

③Java版本>=1.6。

首先以root权限登陆linux系统，执行以下命令查看linux处理器：[root@localhost bin]# uname -aLinux localhost.localdomain 2.6.32-358.el6.x86_64 #1 SMP Tue Jan 29 11:47:41 EST 2013 x86_64 x86_64 x86_64 GNU/Linux执行以下命令查看Python版本：[root@localhost bin]# python --versionPython 2.6.6执行以下命令查看java版本：[root@localhost bin]# java -versionjava version "1.7.0_09-icedtea"2.2mysql数据库安装下载mysql源码，并解压安装。

新建mysql组以及用户：#groupadd mysql#useradd –g mysql mysql；源码包解压：#tar mysql-5.1.50-linux-x86_64-glibc23.tar.gz ；将解压后的源码包放置/usr/local,并改名为mysql；设置目录的所属用户为mysql ：#chown –R mysql /usr/local/mysql ；目录的所属组为mysql ：#chgrp –R mysql /usr/local/mysql ；初始化数据库，一定要在mysql根目录中执行mysql_install_db：#/usr/local/mysql/scripts/mysql_install_db –user=mysql；默认在/user/local/mysql/data设置目录的所属用户为root：#chown –R root/usr/local/mysql；设置目录的所属用户为mysql ：#chown –R mysql /usr/local/mysql/data；复制mysql的配置文件到/etc目录下，并改名为f：cp /usr/local/mysql/support-files/f /etc/f；复制启动脚本到/etc/rc.d/init.d目录下，并命名为mysqld：cp/usr/local/mysql/support-files/mysql.server /etc/rc.d/init.d/mysqld；#service mysqld start 立即启动mysql；设置mysql的密码为：#/usr/local/mysql/bin/mysqladmin –u root password 123456；然后输入密码进入mysql ，退出mysql命令quit:#/usr/local/mysql/bin/mysql –u root –p;若链接不上数据库做以下处理：#vi /etc/sysconf/iptables-A INPUT -m state --state NEW -m tcp -p tcp --dport 3306 -j ACCEPTgrant all privileges on *.* to root@'%' identified by "123";2.3oracle数据库安装一、修改操作系统核心参数在Root用户下执行以下步骤：1）修改用户的SHELL的限制，修改/etc/security/limits.conf文件输入命令：vi /etc/security/limits.conf，按i键进入编辑模式，将下列内容加入该文件。

通过DataX同步数据⾄Elasticsearch 使⽤总结Long值导⼊时精度丢失，字段类型使⽤textsplitPk使⽤ID流⽔号时，导⼊⽆进度0%因Id取最⼩值递加⾄最⼤值，范围间隔⼤空查询较多将数据源的查询时间拆细ES⽇期字段创建需指定格式 yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis⽇期数据导⼊时，text写⼊为⽇期格式，long写⼊为时间戳（数据有精度错误）注意时区问题写⼊时指定时区或对UTC时间戳进⾏转换指定：“2019-03-12T12:12:12.123+0800”转换：东⼋区时间戳 = 3600000*8 + UTC时间戳执⾏命令python /datax/bin/datax.py --jvm=-Xmx4g job.json配置样例job.json{"job": {"setting": {"speed": {"channel": 1}},"content": [{"reader": {...},"writer": {"name": "elasticsearchwriter","parameter": {"endpoint": "http://xxx:9999","accessId": "xxxx","accessKey": "xxxx","index": "test-1","type": "default","cleanup": true,"settings": {"index" :{"number_of_shards": 1, "number_of_replicas": 0}},"discovery": false,"batchSize": 1000,"splitter": ",","column": [{"name": "pk", "type": "id"},{ "name": "col_ip","type": "ip" },{ "name": "col_double","type": "double" },{ "name": "col_long","type": "long" },{ "name": "col_integer","type": "integer" },{ "name": "col_keyword", "type": "keyword" },{ "name": "col_text", "type": "text", "analyzer": "ik_max_word"},{ "name": "col_geo_point", "type": "geo_point" },{ "name": "col_date", "type": "date", "format": "yyyy-MM-dd HH:mm:ss"},{ "name": "col_nested1", "type": "nested" },{ "name": "col_nested2", "type": "nested" },{ "name": "col_object1", "type": "object" },{ "name": "col_object2", "type": "object" },{ "name": "col_integer_array", "type":"integer", "array":true},{ "name": "col_geo_shape", "type":"geo_shape", "tree": "quadtree", "precision": "10m"} ]}}}]}}参数说明endpoint描述：ElasticSearch的连接地址必选：是默认值：⽆accessId描述：http auth中的user必选：否默认值：空accessKey描述：http auth中的password必选：否默认值：空index描述：elasticsearch中的index名必选：是默认值：⽆type描述：elasticsearch中index的type名必选：否默认值：index名cleanup描述：是否删除原表必选：否默认值：falsebatchSize描述：每次批量数据的条数必选：否默认值：1000trySize描述：失败后重试的次数必选：否默认值：30timeout描述：客户端超时时间必选：否默认值：600000discovery描述：启⽤节点发现将(轮询)并定期更新客户机中的服务器列表。