Datax mysql8插件开发

其实mysql8和mysql的区别几乎只是改了驱动而已, 过程一模一样的,那么开发步骤也是很简单, 几乎都是复制mysql的插件就可以了(如果业务不是像我一样又要mysql5.7又要mysql8的话直接改插件就可以了,不需要新建插件)

这篇文章只用mysql8reader举例, mysql8的writer大家可以自己写

1. 新建module

ps: 我这里冒红只是因为我已经建立过了

2. mysql8配置文件处理

对于一个插件来说, 需要配置的文件有这么几个:

1. mysql8reader/src/main/resources/{json文件}: 负责生成jar包和json模板
2. pom.xml: 负责maven编译,配置依赖项
3. mysql8reader/src/main/assembly/package.xml: 用来选择文件生成的地址(assembly文件夹需要自己创建)

依次的代码为(可以从其他的插件中复制后进行一定的修改):

1. plugin.json和plugin_job_template.json

// plugin.json
{
    "name": "mysql8reader","class": "com.alibaba.datax.plugin.reader.mysql8reader.Mysql8Reader","description": "useScene: prod. mechanism: Jdbc connection using the database, execute select sql, retrieve data from the ResultSet. warn: The more you know about the database, the less problems you encounter.","developer": "alibaba"
}

// plugin_job_template.json
{
    "name": "mysql8reader","parameter": {
    "username": "","password": "","column": [],"connection": [{
    "jdbcUrl": [],"table": []}],"where": ""}
}

2. pom.xml(位于mysql8reader下)

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>com.alibaba.datax</groupId><artifactId>datax-all</artifactId><version>0.0.1-SNAPSHOT</version></parent><artifactId>mysql8reader</artifactId><name>mysql8reader</name><packaging>jar</packaging><dependencies><dependency><groupId>com.alibaba.datax</groupId><artifactId>datax-common</artifactId><version>${datax-project-version}</version><exclusions><exclusion><artifactId>slf4j-log4j12</artifactId><groupId>org.slf4j</groupId></exclusion></exclusions></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId></dependency><dependency><groupId>ch.qos.logback</groupId><artifactId>logback-classic</artifactId></dependency><dependency><groupId>com.alibaba.datax</groupId><artifactId>plugin-rdbms-util</artifactId><version>${datax-project-version}</version></dependency><!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java --><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.21</version></dependency></dependencies><build><plugins><!-- compiler plugin --><plugin><artifactId>maven-compiler-plugin</artifactId><configuration><source>${jdk-version}</source><target>${jdk-version}</target><encoding>${project-sourceEncoding}</encoding></configuration></plugin><!-- assembly plugin --><plugin><artifactId>maven-assembly-plugin</artifactId><configuration><descriptors><descriptor>src/main/assembly/package.xml</descriptor></descriptors><finalName>datax</finalName></configuration><executions><execution><id>dwzip</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins></build></project>

3. package.xml(位于mysql8reader/src/main/assembly/下, assembly文件夹需要手动创建)

<assemblyxmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd"><id></id><formats><format>dir</format></formats><includeBaseDirectory>false</includeBaseDirectory><fileSets><fileSet><directory>src/main/resources</directory><includes><include>plugin.json</include><include>plugin_job_template.json</include></includes><outputDirectory>plugin/reader/mysql8reader</outputDirectory></fileSet><fileSet><directory>target/</directory><includes><include>mysql8reader-0.0.1-SNAPSHOT.jar</include></includes><outputDirectory>plugin/reader/mysql8reader</outputDirectory></fileSet></fileSets><dependencySets><dependencySet><useProjectArtifact>false</useProjectArtifact><outputDirectory>plugin/reader/mysql8reader/libs</outputDirectory><scope>runtime</scope></dependencySet></dependencySets>
</assembly>

3. Java代码:

在mysql8reader/src/main/java下创建一个package,命名为:
“com.alibaba.datax.plugin.reader.mysql8reader”

然后在这个package下创建一个class,命名为:
“Mysql8Reader”

由于mysql8的流程和mysql差不多所以代码层面没有什么要改的东西

package com.alibaba.datax.plugin.reader.mysql8reader;import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.spi.Reader;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader;
import com.alibaba.datax.plugin.rdbms.reader.Constant;
import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.util.List;public class Mysql8Reader extends Reader{
    private static final DataBaseType DATABASE_TYPE = DataBaseType.MySql;public static class Job extends Reader.Job{
    private static final Logger Log = LoggerFactory.getLogger(Job.class);private Configuration originalConfig = null;private CommonRdbmsReader.Job commonRdbmsReaderJob;@Overridepublic void init(){
    this.originalConfig = super.getPluginJobConf();Integer userConfigedFetchSize = this.originalConfig.getInt(Constant.FETCH_SIZE);if (userConfigedFetchSize != null) {
    Log.warn("对 mysqlreader 不需要配置 fetchSize, mysqlreader 将会忽略这项配置. 如果您不想再看到此警告,请去除fetchSize 配置.");}this.originalConfig.set(Constant.FETCH_SIZE, Integer.MIN_VALUE);this.commonRdbmsReaderJob = new CommonRdbmsReader.Job(DATABASE_TYPE);this.commonRdbmsReaderJob.init(this.originalConfig);}@Overridepublic void preCheck(){
    init();this.commonRdbmsReaderJob.preCheck(this.originalConfig, DATABASE_TYPE);}@Overridepublic List<Configuration> split(int adviceNumber){
    return this.commonRdbmsReaderJob.split(this.originalConfig, adviceNumber);}@Overridepublic void post(){
    this.commonRdbmsReaderJob.post(this.originalConfig);}@Overridepublic void destroy(){
    this.commonRdbmsReaderJob.destroy(this.originalConfig);}}public static class Task extends Reader.Task{
    private Configuration readerSliceConfig;private CommonRdbmsReader.Task commonRdbmsReaderTask;@Overridepublic void init(){
    this.readerSliceConfig = super.getPluginJobConf();this.commonRdbmsReaderTask = new CommonRdbmsReader.Task(DATABASE_TYPE, super.getTaskGroupId(), super.getTaskId());this.commonRdbmsReaderTask.init(this.readerSliceConfig);}@Overridepublic void startRead(RecordSender recordSender){
    int fetchSize = this.readerSliceConfig.getInt(Constant.FETCH_SIZE);this.commonRdbmsReaderTask.startRead(this.readerSliceConfig, recordSender,super.getTaskPluginCollector(), fetchSize);}@Overridepublic void post(){
    this.commonRdbmsReaderTask.post(this.readerSliceConfig);}@Overridepublic void destroy(){
    this.commonRdbmsReaderTask.destroy(this.readerSliceConfig);}}}

4.构建项目的配置文件

修改package.xml和pom.xml(位于Datax/下)

1. 修改package.xml, 添加mysql8reader的配置项
   
2. 修改pom.xml, 添加对应的配置
   

5. 编译运行测试

1. 执行maven命令进行编译

 mvn -U clean package assembly:assembly -Dmaven.test.skip=true

编译好的结果

2. 构建json文件进行测试

{
    "job": {
    "setting": {
    "speed": {
    "channel": "5"},"errorLimit": {
    "record": 1}},"content": [{
    "reader": {
    "name": "mysql8reader","parameter": {
    "username": "username","password": "password","column": ["`id`","`name`"],"connection": [{
    "table": ["`tablename"],"jdbcUrl": ["jdbc:mysql://ip:port/database_name?useUnicode=true&characterEncoding=utf8"]}]}},"writer": {
    "name": "mysqlwriter","parameter": {
    "writeMode": "insert","username": "username","password": "password","column": ["`id`","`name`"],"session": ["set session sql_mode='ANSI'"],"preSql": [],"connection": [{
    "jdbcUrl": "jdbc:mysql://ip:port/database_name?characterEncoding=utf8","table": ["`tablename`"]}]}}}]}
}

3. 运行查看结果

python datax.py mysql8test.json

4. 大功告成