java 简单datax使用记录（笔记）

待我称王封你为后i 2023-02-20 02:46 18阅读 0赞

### java使用datax ###

*   *  启动：
     *  Datax 同步增量配置
     *  修改job.json 文件最下方增加：
     *  脚本：

Datax 是阿里离线数据同步工具，实现包括MySQL、Oracle、SqlServer、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、DRDS 等各种异构数据源之间高效的数据同步功能。  
DataX本身作为数据同步框架，将不同数据源的同步抽象为从源头数据源读取数据的Reader插件，以及向目标端写入数据的Writer插件，理论上DataX框架可以支持任意数据源类型的数据同步工作。同时DataX插件体系作为一套生态系统, 每接入一套新数据源该新加入的数据源即可实现和现有的数据源互通。  
开源地址：https://github.com/alibaba/DataX/  
Datax 核心分为：  
Reader：数据采集模块，负责从源采集数据  
　　　　Writer：数据写入模块，负责写入目标库  
　　　　Framework：数据传输通道，负责处理数据缓冲等  
使用：  
需要安装：  
• JDK(1.8以上，推荐1.8)  
• Python(推荐Python2.6.X)  
• Apache Maven 3.x (Compile DataX)

java 使用参考博客 ：https://blog.csdn.net/chenzhong555/article/details/86510936  
1、下载datax 源码  
2、将你需要的内容打包成jar包 引入你的项目中，由于是开源项目你可以先进行修改源码后再进行打包

注意： 所有需要的库相关jar datax 全部都需要引入进来，mysql 数据库需要查看你的数据库版本，若链接错误可以试着将mysql-connector-java-8.0.15.jar，拷贝到\{datax.home\}\\plugin\\writer\\mysqlwriter\\libs目录下即可，reader读取同理。

一、  
下载datax 工具  
安装包链接：http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz  
解压后修改 /job/job.sjon文件 该文件为你的数据同步内容。  
重要：job.json文件中不允许有注释存在且格式必须正确

{ 
    	"job": { 
    		"setting": { 
    			"speed": { 
    				"channel": 2,		//同时执行并发数
    				"record":200,		
    				"byte": 2242880,	
    				"batchSize":2048	//单次执行条数 
    },
    			"errorLimit": { 
    				"record": 10,	//允许每次执行错误条数
    				"percentage": 0.1	//允许每次执行错误比例
    			}
    		},
    		"content": [{ 
    			"reader": { 	//reader为源数据配置 
    //此处名称不要随便写，如我这里是sqlserver数据库则填写的sqlserverreader。Mysql 则可以为mysqlreader此处可以参考源码中每个数据库名称。 
    				"name": "sqlserverreader",	 
    				"parameter": { 
    					"sliceRecordCount": 10,
    					"connection": [{ 
    						"querySql": [	//此处查询出的结果为要像目标库同步的数据
    						     "select
    						     id ,
    						     BabyNumber ,
    						     Child_No ,
    						     BabyName ,
    						     EnBabyName 
    						     from InitNewBaby where 1=1"
    						],
    			//数据库url及库名
    			"jdbcUrl":["jdbc:sqlserver://127.0.0.1:1433;DatabaseName=jk_database_tlbak"]	
    					}],
    					"username": "babyplan",	//数据库登录名
    					"password": "123456"	//数据库密码
    				}
    			},
    			"writer": { 	//此处为目标库配置
    				"name": "mysqlwriter", 	//mysql 数据库
    				"parameter": { 
    					"column": [	//要同步的字段
    						"id",
    					    "BabyNumber",
    					    "Child_No",
    					    "BabyName",
    					    "EnBabyName",
    					],
    					"connection": [{ 	//配置jdbc url+数据库名称
    						"jdbcUrl":jdbc:mysql://127.0.0.1:3306/mysqldb?serverTimezone=Asia/Shanghai&characterEncoding=UTF-8",							
    						"table": [
    							"InitNewBaby"		//表名称
    						]
    					}],
    					"password": "root",		//数据库密码
    					"username": "root"		//用户名
    				}
    			}
    		}]
    	}
    }

## 启动： ##

启动方式有两种，一种为java 启动，一种为命令行直接启动

Java 启动：  
Java 启动时首先需要在项目中引入datax 相关jar，其实就是datax源码，然后自己maven 打包引入到项目中。  
源码地址：https://github.com/alibaba/DataX/  
源码中有很多项目模块，需要哪个就下载哪个。  
如：  
当前项目源数据为sqlserver 数据库，目标数据库为mysql 那么就需要打包源码中sqlserverreader和 mysqlwriter 当然还有必须有的 common、core、transformer 这三个。你需要将打包好的jar 放到你项目的maven 库中，然后由你在项目pom 文件中进行引用。  
注意： 所有需要的库相关jar datax 全部都需要引入进来，mysql 数据库需要查看你的数据库版本，若链接错误可以试着将mysql-connector-java-8.0.15.jar，拷贝到datax工具中 datax\\plugin\\writer\\mysqlwriter\\libs目录下即可，reader读取同理。  
注意你jar 缩放的路径

相关pom

<dependency>
                <groupId>com.alibaba.datax</groupId>
                <artifactId>datax-core</artifactId>
                <version>0.0.1-SNAPSHOT</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba.datax</groupId>
                <artifactId>sqlserverreader</artifactId>
                <version>0.0.1-SNAPSHOT</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba.datax</groupId>
                <artifactId>sqlserverwriter</artifactId>
                <version>0.0.1-SNAPSHOT</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba.datax</groupId>
                <artifactId>mysqlwriter</artifactId>
                <version>0.0.1-SNAPSHOT</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba.datax</groupId>
                <artifactId>mysqlreader</artifactId>
                <version>0.0.1-SNAPSHOT</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba.datax</groupId>
                <artifactId>plugin-rdbms-util</artifactId>
                <version>0.0.1-SNAPSHOT</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba.datax</groupId>
                <artifactId>datax-common</artifactId>
                <version>0.0.1-SNAPSHOT</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba.datax</groupId>
                <artifactId>datax-transformer</artifactId>
                <version>0.0.1-SNAPSHOT</version>
            </dependency>

Java 启动方法：

private static final String DATAX_PATH = "E:/soft/datax/datax"; //datax 所在路径
    
        /** * 参数说明: * @param dataxPath 是datax工具的路径 * @param jsonPath 是配置同步数据库json文件的路径 * @throws Throwable * System.setProperty(“datax.home”, dataxPath); 设置datax.home环境变量， * 在系统启动时会通过datax.home环境变量获取配置文件和read、 * writer数据量插件。String[] datxArgs = {"-job", jsonPath, “-mode”, “standalone”, “-jobid”, “-1”}; * 定义构造参数数组，在系统启动时会用到。 * 通过方法就可以使用Java的方式启动DataX */
        public static void start(String dataxPath, String jsonPath) throws Throwable { 
            System.setProperty("datax.home", dataxPath);
            System.setProperty("now", new SimpleDateFormat("yyyy/MM/dd-HH:mm:ss:SSS").format(new Date()));// 替换job中的占位符
            System.setProperty("user.timezone","GMT+8");    //设置时区 https://blog.csdn.net/Shadow_Light/article/details/102615364
            String[] datxArgs = { "-job", jsonPath, "-mode", "standalone", "-jobid", "-1"};
    // Engine.getResult();
        }
    	//这里其实可以根据条件 调用多个 job.json 文件
        public static void main(String[] args) throws Throwable { 
            start(DATAX_PATH, DATAX_PATH +"/job/job_insert.json");
        }

linux 启动命令  
确保服务器已经安装Python

python /opt/datax/bin/datax.py /opt/datax/job/job.json

//注意修改成自己datax和job.json文件所在路径及名称

注：若导入数据时间不对可能是时区问题。  
1、查看服务器时区

## Datax 同步增量配置 ##

同步增量思路，先读取一个最大值（ID 或时间），然后根据最大值在源库查询时先过滤。
    首先创建一个可以读取最大值得.json文件。
    配置如下：
    		注意json 文件中不允许有注释

{ 
        "job": { 
            "content": [	
                { 
                    "reader": { 	//当前配置为要读取的最大值（同步数据时源数据的筛选条件（参数））
                        "name": "mysqlreader",	
                        "parameter": { 
                            "connection": [
                                { 
                                    "jdbcUrl": ["jdbc:mysql://127.0.0.1:8030/saitron_babyplan"],
                                    "querySql": [//按照最大时间进行增量同步，此处查询出目标数据库中已经同步好的最大时间
                                        "SELECT max(filetime) FROM InitNewBaby"
                                    ]
                                }
                            ],
                            "password": "root",
                            "username": "root"
                        }
                    },
                    "writer": { 
                        "name": "txtfilewriter",	//名称
                        "parameter": { 
                           "dateFormat": "yyyy-MM-dd HH:mm:ss",//查询出最大时间保存在的文件名称，保存在max_timestamp_result文件下
    						"fileName": "max_timestamp_result",	                      
    						"fileFormat": "csv",	//为csv格式
                            "path": "/opt/datax/bin", //文件存储的路径，在/opt/datax/bin下
                            "writeMode": "truncate"	
                        }
                    }
                }
            ],
            "setting": { 
    		"speed": { 
    			"channel": 2
    		}
    	}
        }
    }

## 修改job.json 文件最下方增加： ##

{ 
    	"job": { 
    		"setting": { 
    			"speed": { 
    				"channel": 2,		//同时执行并发数
    				"record":200,		
    				"byte": 2242880,	
    				"batchSize":2048	//单次执行条数 
    },
    			"errorLimit": { 
    				"record": 10,	//允许每次执行错误条数
    				"percentage": 0.1	//允许每次执行错误比例
    			}
    		},
    		"content": [{ 
    			"reader": { 	//reader为源数据配置 
    //此处名称不要随便写，如我这里是sqlserver数据库则填写的sqlserverreader。Mysql 则可以为mysqlreader此处可以参考源码中每个数据库名称。 
    				"name": "sqlserverreader",	 
    				"parameter": { 
    					"sliceRecordCount": 10,
    					"connection": [{ 
    						"querySql": [	//此处查询出的结果为要像目标库同步的数据
    						     "select
    						     id ,
    						     BabyNumber ,
    						     Child_No ,
    						     BabyName ,
    						     EnBabyName 
    						     from InitNewBaby where 1=1"
    						],
    			//数据库url及库名
    			"jdbcUrl":["jdbc:sqlserver://127.0.0.1:1433;DatabaseName=jk_database_tlbak"]	
    					}],
    					"username": "babyplan",	//数据库登录名
    					"password": "admin@2020"	//数据库密码
    				}
    			},
    			"writer": { 	//此处为目标库配置
    				"name": "mysqlwriter", 	//mysql 数据库
    				"parameter": { 
    					"column": [	//要同步的字段
    						"id",
    					    "BabyNumber",
    					    "Child_No",
    					    "BabyName",
    					    "EnBabyName",
    					],
    					"connection": [{ 	//配置jdbc url+数据库名称
    						"jdbcUrl": "jdbc:mysql://127.0.0.1:9030/saitron_babyplan",
    						"table": [
    							"InitNewBaby"		//表名称
    						]
    					}],
    					"password": "root",		//数据库密码
    					"username": "root"		//用户名
    				}
    				"transformer": [ //此处为新增内容
    				    { 
    					"name": "dx_filter",
    					"parameter": 
    					    { 
    					    "columnIndex":1,
    					    "paras":["<","timestamp"]	//此处设置timestamp 用于在脚本中替换成在目标库中查询到的筛选条件（id或日期）
    					    }  
    				    }
    				]
    			}
    		}]
    	}
    }

## 脚本： ##

将相关内容替换成你自己的

#JAVA_HOME=/usr/java/jdk1.8.0_152
    #export PATH=$PATH:$JAVA_HOME/bin
    ### every exit != 0 fails the script
    set –e
    #运行读取max_timestamp2csv.json 文件获取最大筛选条件（id、时间）
    python /opt/datax/bin/datax.py /opt/datax/job/scripts/max_timestamp2csv.json
    # $?是shell变量,表示"最后一次执行命令"的退出状态.0为成功,非0为失败, -ne 为不等于
    if [ $? -ne 0 ]; then
      echo "minute_data_sync.sh error, can not get max_time from target db!"
      exit 1
    fi
    # 找到 DataX 写入的文本文件，并将内容读取到一个变量中（该文件为max_timestamp2csv.json查询出的#最大时间）
    RESULT_FILE=`ls /opt/datax/bin/max_timestamp_result_*`
    MAX_TIME=`cat $RESULT_FILE`
    echo "$RESULT_FILE $MAX_TIME"
    # 如果最大时间戳不为 null 的话， 修改全部同步的配置，进行增量更新；
    if [ "$MAX_TIME" != "null" ]; then
      # 设置增量更新过滤条件
      WHERE="filetime > '$MAX_TIME'"
      # 将incre_update.json 配置文件中1=1条件替换为上次同步数据的最大时间戳
      sed "s/1=1/$WHERE/g" /opt/datax/job/incre_update.json > /opt/datax/job/incre_update_tmp.json
      #echo "增量更新"
      python /opt/datax/bin/datax.py /opt/datax/job/incre_update_tmp.json
      # 删除临时文件
      rm ./incre_update_tmp.json
    else
      # echo "全部更新"
      python /opt/datax/bin/datax.py /opt/datax/job/job.json
    fi