Skip to content

DataX插件开发宝典

binaryWorld edited this page Jun 24, 2016 · 3 revisions

DataX插件开发宝典

本文面向DataX插件开发人员,尝试尽可能全面地阐述开发一个DataX插件所经过的历程,力求消除开发者的困惑,让插件开发变得简单。

一、开发之前

路走对了,就不怕远。✓ 路走远了,就不管对不对。✕

当你打开这篇文档,想必已经不用在此解释什么是DataX了。那下一个问题便是:

DataX为什么要使用插件机制?

从设计之初,DataX就把异构数据源同步作为自身的使命,为了应对不同数据源的差异、同时提供一致的同步原语和扩展能力,DataX自然而然地采用了框架 + 插件 的模式:

  • 插件只需关心数据的读取或者写入本身。
  • 而同步的共性问题,比如:类型转换、性能、统计,则交由框架来处理。

作为插件开发人员,则需要关注两个问题:

  1. 数据源本身的读写数据正确性。
  2. 如何与框架沟通、合理正确地使用框架。

二、插件视角看框架

逻辑执行模型

插件开发者不用关心太多,基本只需要关注特定系统读和写,以及自己的代码在逻辑上是怎样被执行的,哪一个方法是在什么时候被调用的。在此之前,需要明确以下概念:

  • Job: Job是DataX用以描述从一个源头到一个目的端的同步作业,是DataX数据同步的最小业务单元。比如:从一张mysql的表同步到odps的一个表的特定分区。
  • Task: Task是为最大化而把Job拆分得到的最小执行单元。比如:读一张有1024个分表的mysql分库分表的Job,拆分成1024个读Task,用若干个并发执行。
  • TaskGroup: 描述的是一组Task集合。在同一个TaskGroupContainer执行下的Task集合称之为TaskGroup
  • JobContainer: Job执行器,负责Job全局拆分、调度、前置语句和后置语句等工作的工作单元。类似Yarn中的JobTracker
  • TaskGroupContainer: TaskGroup执行器,负责执行一组Task的工作单元,类似Yarn中的TaskTracker。

上述这些概念的逻辑结构如下:

Job-Task

简而言之, Job拆分成Task,在分别在框架提供的容器中执行,插件只需要实现JobTask两部分逻辑

三、如何开发新的插件

3.1 设计插件参数

配置文件的设计是插件开发的第一步!

任务配置中readerwriterparameter部分是插件的配置参数;如MysqlReader插件的参数如下:

{
   "job": {
       "setting": {
           "speed": {
               "channel": 5
           }
       },
       "content": [
           {
               "reader": {
                   "name": "mysqlreader",
                   "parameter": {
                       "username": "root",
                       "password": "root",
                       "column": [],
                       "where": "",
                       "session": {},
                       "connection": [
                           {
                               "table": [],
                               "jdbcUrl": []
                           }
                       ]
                   }
               },
               "writer": {}
           }
       ]
   }
}

插件的配置参数应当遵循以下原则:

  • 驼峰命名:所有配置项采用驼峰命名法,首字母小写,单词首字母大写。

  • 正交原则:配置项必须正交,功能没有重复,没有潜规则。

  • 富类型:合理使用json的类型,减少无谓的处理逻辑,减少出错的可能。

    • 使用正确的数据类型。比如,bool类型的值使用true/false,而非"yes"/"true"/0等。
    • 合理使用集合类型,比如,用数组替代有分隔符的字符串。
  • 类似通用:遵守同一类型的插件的习惯,比如关系型数据库的connection参数都是如下结构:

    {
      "connection": [
        {
          "table": [
            "table_1",
            "table_2"
          ],
          "jdbcUrl": [
            "jdbc:mysql://127.0.0.1:3306/database_1",
            "jdbc:mysql://127.0.0.2:3306/database_1_slave"
          ]
        },
        {
          "table": [
            "table_3",
            "table_4"
          ],
          "jdbcUrl": [
            "jdbc:mysql://127.0.0.3:3306/database_2",
            "jdbc:mysql://127.0.0.4:3306/database_2_slave"
          ]
        }
      ]
    }
    

3.2 扩展ReaderWriter,实现Reader插件或Writer插件

插件的入口类必须扩展ReaderWriter抽象类,并且实现分别实现JobTask两个内部抽象类,JobTask的实现必须是 内部类 的形式,原因见 加载原理 一节。以Reader为例:MyReader

public class MyReader extends Reader {
    public static class Job extends Reader.Job {

        @Override
        public void init() {
        }
		
		@Override
		public void prepare() {
        }

        @Override
        public List<Configuration> split(int adviceNumber) {
            return null;
        }

        @Override
        public void post() {
        }

        @Override
        public void destroy() {
        }

    }

    public static class Task extends Reader.Task {

        @Override
        public void init() {
        }
		
		@Override
		public void prepare() {
        }

        @Override
        public void startRead(RecordSender recordSender) {
        }

        @Override
        public void post() {
        }

        @Override
        public void destroy() {
        }
    }
}

Job接口功能如下:

  • init: Job对象初始化工作,测试可以通过super.getPluginJobConf()获取与本插件相关的配置。读插件获得配置中reader部分,写插件获得writer部分。
  • prepare: 全局准备工作,比如odpswriter清空目标表。
  • split: 拆分Task。参数adviceNumber框架建议的拆分数,一般是运行时所配置的并发度。值返回的是Task的配置列表。
  • post: 全局的后置工作,比如mysqlwriter同步完影子表后的rename操作。
  • destroy: Job对象自身的销毁工作。

Task接口功能如下:

  • init:Task对象的初始化。此时可以通过super.getPluginJobConf()获取与本Task相关的配置。这里的配置是Jobsplit方法返回的配置列表中的其中一个。
  • prepare:局部的准备工作。
  • startRead: 从数据源读数据,写入到RecordSender中。RecordSender会把数据写入连接Reader和Writer的缓存队列。
  • startWrite:从RecordReceiver中读取数据,写入目标数据源。RecordReceiver中的数据来自Reader和Writer之间的缓存队列。
  • post: 局部的后置工作。
  • destroy: Task象自身的销毁工作。

需要注意的是:

  • JobTask之间一定不能有共享变量,因为分布式运行时不能保证共享变量会被正确初始化。两者之间只能通过配置文件进行依赖。
  • preparepostJobTask中都存在,插件需要根据实际情况确定在什么地方执行操作。

框架按照如下的顺序执行JobTask的接口:

DataXReaderWriter (2)

上图中,黄色表示Job部分的执行阶段,蓝色表示Task部分的执行阶段,绿色表示框架执行阶段。

相关类关系如下:

DataX

3.3 定义插件相关信息

代码写好了,有没有想过框架是怎么找到插件的入口类的?框架是如何加载插件的呢?

在每个插件的项目中,都有一个plugin.json文件,这个文件定义了插件的相关信息,包括入口类。例如:

{
    "name": "myreader",
    "class": "com.alibaba.datax.plugin.reader.myreader.MyReader",
    "description": "Use Jdbc connect to database, execute insert sql.",
    "developer": "binaryworld"
}
  • name: 插件名称,大小写敏感。框架根据用户在配置文件中指定的名称来搜寻插件。 十分重要
  • class: 入口类的全限定名称,框架通过反射穿件入口类的实例。十分重要
  • description: 描述信息。
  • developer: 开发人员。

3.4 在 DataX 框架中注册该插件

  • 步骤1:在 DataX3的目录下的 pom.xml 中添加本插件的模块名称 模块名称artifactId
 <modules>
        <module>myreader</module>     
 </modules>       
  • 步骤2:在 DataX3的目录下的 package.xml 中添加本插件的打包内容,以让 DataX 能够把该插件纳入整个插件体系。
 <fileSet>
        <directory>myreader(插件名称pluginName)/target/datax/</directory>
        <includes>
            <include>**/*.*</include>
        </includes>
        <outputDirectory>datax</outputDirectory>
</fileSet>     

3.5 打包发布

DataX使用assembly打包,assembly的使用方法请咨询谷哥或者度娘。打包命令如下:

mvn clean package -DskipTests assembly:assembly

DataX插件需要遵循统一的目录结构:

${DATAX_HOME}
|-- bin       
|   `-- datax.py
|-- conf
|   |-- core.json
|   `-- logback.xml
|-- lib
|   `-- datax-core-dependencies.jar
`-- plugin
    |-- reader
    |   `-- myreader
    |       |-- libs
    |       |   `-- my-reader-plugin-dependencies.jar
    |       |-- myreader-0.0.1-SNAPSHOT.jar
    |       `-- plugin.json
    `-- writer
        |-- mywriter
        |   |-- libs
        |   |   `-- my-writer-plugin-dependencies.jar
        |   |-- mywriter-0.0.1-SNAPSHOT.jar
        |   `-- plugin.json
        |-- oceanbasewriter
        `-- odpswriter
  • ${DATAX_HOME}/bin: 可执行程序目录。
  • ${DATAX_HOME}/conf: 框架配置目录。
  • ${DATAX_HOME}/lib: 框架依赖库目录。
  • ${DATAX_HOME}/plugin: 插件目录。

插件目录分为readerwriter子目录,读写插件分别存放。插件目录规范如下:

  • ${PLUGIN_HOME}/libs: 插件的依赖库。
  • ${PLUGIN_HOME}/plugin-name-version.jar: 插件本身的jar。
  • ${PLUGIN_HOME}/plugin.json: 插件描述文件。

尽管框架加载插件时,会把${PLUGIN_HOME}下所有的jar放到classpath,但还是推荐依赖库的jar和插件本身的jar分开存放。

注意: 插件的目录名字必须和plugin.json中定义的插件名称一致。

配置文件

DataX使用json作为配置文件的格式。一个典型的DataX任务配置如下:

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "odpsreader",
          "parameter": {
            "accessKey": "xxx",
            "accessId": "xxx",
            "column": [""],
            "isCompress": "false",
            "odpsServer": "xxx",
            "partition": [
              "xx"
            ],
            "project": "decdm",
            "table": "",
            "tunnelServer": ""
          }
        },
        "writer": {
          "name": "oraclewriter",
          "parameter": {
            "username": "user",
            "password": "",
            "column": ["*"],
            "connection": [
              {
                "jdbcUrl": "",
                "table": [
                  ""
                ]
              }
            ]
          }
        }
      }
    ]
  }
}

DataX框架有core.json配置文件,指定了框架的默认行为。任务的配置里头可以指定框架中已经存在的配置项,而且具有更高的优先级,会覆盖core.json中的默认值。

配置中job.content.reader.parameter的value部分会传给Reader.Jobjob.content.writer.parameter的value部分会传给Writer.JobReader.JobWriter.Job可以通过super.getPluginJobConf()来获取。

DataX框架支持对特定的配置项进行RSA加密,例子中以*开头的项目便是加密后的值。 配置项加密解密过程对插件是透明,插件仍然以不带*的key来查询配置和操作配置项

如何使用Configuration

为了简化对json的操作,DataX提供了简单的DSL配合Configuration类使用。

Configuration提供了常见的get, 带类型get带默认值getset等读写配置项的操作,以及clone, toJSON等方法。配置项读写操作都需要传入一个path做为参数,这个path就是DataX定义的DSL。语法有两条:

  1. 子map用.key表示,path的第一个点省略。
  2. 数组元素用[index]表示。

比如操作如下json:

{
  "a": {
    "b": {
      "c": 2
    },
    "f": [
      1,
      2,
      {
        "g": true,
        "h": false
      },
      4
    ]
  },
  "x": 4
}

比如调用configuration.get(path)方法,当path为如下值的时候得到的结果为:

  • x4
  • a.b.c2
  • a.b.c.dnull
  • a.b.f[0]1
  • a.b.f[2].gtrue

注意,因为插件看到的配置只是整个配置的一部分。使用Configuration对象时,需要注意当前的根路径是什么。

更多Configuration的操作请参考ConfigurationTest.java

插件数据传输

跟一般的生产者-消费者模式一样,Reader插件和Writer插件之间也是通过channel来实现数据的传输的。channel可以是内存的,也可能是持久化的,插件不必关心。插件通过RecordSenderchannel写入数据,通过RecordReceiverchannel读取数据。

channel中的一条数据为一个Record的对象,Record中可以放多个Column对象,这可以简单理解为数据库中的记录和列。

Record有如下方法:

public interface Record {
    // 加入一个列,放在最后的位置
    void addColumn(Column column);
    // 在指定下标处放置一个列
    void setColumn(int i, final Column column);
    // 获取一个列
    Column getColumn(int i);
    // 转换为json String
    String toString();
    // 获取总列数
    int getColumnNumber();
    // 计算整条记录在内存中占用的字节数
    int getByteSize();
}

因为Record是一个接口,Reader插件首先调用RecordSender.createRecord()创建一个Record实例,然后把Column一个个添加到Record中。

Writer插件调用RecordReceiver.getFromReader()方法获取Record,然后把Column遍历出来,写入目标存储中。当Reader尚未退出,传输还在进行时,如果暂时没有数据RecordReceiver.getFromReader()方法会阻塞直到有数据。如果传输已经结束,会返回nullWriter插件可以据此判断是否结束startWrite方法。

Column的构造和操作,我们在《类型转换》一节介绍。

类型转换

为了规范源端和目的端类型转换操作,保证数据不失真,DataX支持六种内部数据类型:

  • Long:定点数(Int、Short、Long、BigInteger等)。
  • Double:浮点数(Float、Double、BigDecimal(无限精度)等)。
  • String:字符串类型,底层不限长,使用通用字符集(Unicode)。
  • Date:日期类型。
  • Bool:布尔值。
  • Bytes:二进制,可以存放诸如MP3等非结构化数据。

对应地,有DateColumnLongColumnDoubleColumnBytesColumnStringColumnBoolColumn六种Column的实现。

Column除了提供数据相关的方法外,还提供一系列以as开头的数据类型转换转换方法。

Columns

DataX的内部类型在实现上会选用不同的java类型:

内部类型 实现类型 备注
Date java.util.Date
Long java.math.BigInteger 使用无限精度的大整数,保证不失真
Double java.lang.String 用String表示,保证不失真
Bytes byte[]
String java.lang.String
Bool java.lang.Boolean

类型之间相互转换的关系如下:

from\to Date Long Double Bytes String Bool
Date - 使用毫秒时间戳 不支持 不支持 使用系统配置的date/time/datetime格式转换 不支持
Long 作为毫秒时间戳构造Date - BigInteger转为BigDecimal,然后BigDecimal.doubleValue() 不支持 BigInteger.toString() 0为false,否则true
Double 不支持 内部String构造BigDecimal,然后BigDecimal.longValue() - 不支持 直接返回内部String
Bytes 不支持 不支持 不支持 - 按照common.column.encoding配置的编码转换为String,默认utf-8 不支持
String 按照配置的date/time/datetime/extra格式解析 用String构造BigDecimal,然后取longValue() 用String构造BigDecimal,然后取doubleValue(),会正确处理NaN/Infinity/-Infinity 按照common.column.encoding配置的编码转换为byte[],默认utf-8 - "true"为true, "false"为false,大小写不敏感。其他字符串不支持
Bool 不支持 true1L,否则0L true1.0,否则0.0 不支持 -

脏数据处理

什么是脏数据?

目前主要有三类脏数据:

  1. Reader读到不支持的类型、不合法的值。
  2. 不支持的类型转换,比如:Bytes转换为Date
  3. 写入目标端失败,比如:写mysql整型长度超长。

如何处理脏数据

Reader.TaskWriter.Task中,通过AbstractTaskPlugin.getPluginCollector()可以拿到一个TaskPluginCollector,它提供了一系列collectDirtyRecord的方法。当脏数据出现时,只需要调用合适的collectDirtyRecord方法,把被认为是脏数据的Record传入即可。

用户可以在任务的配置中指定脏数据限制条数或者百分比限制,当脏数据超出限制时,框架会结束同步任务,退出。插件需要保证脏数据都被收集到,其他工作交给框架就好。

加载原理

  1. 框架扫描plugin/readerplugin/writer目录,加载每个插件的plugin.json文件。
  2. plugin.json文件中name为key,索引所有的插件配置。如果发现重名的插件,框架会异常退出。
  3. 用户在插件中在reader/writer配置的name字段指定插件名字。框架根据插件的类型(reader/writer)和插件名称去插件的路径下扫描所有的jar,加入classpath
  4. 根据插件配置中定义的入口类,框架通过反射实例化对应的JobTask对象。

四、Last but not Least

文档是工程师的良知。

每个插件都必须在DataX官方wiki中有一篇文档,文档需要包括但不限于以下内容:

  1. 快速介绍:介绍插件的使用场景,特点等。
  2. 实现原理:介绍插件实现的底层原理,比如mysqlwriter通过insert intoreplace into来实现插入,tair插件通过tair客户端实现写入。
  3. 配置说明
    • 给出典型场景下的同步任务的json配置文件。
    • 介绍每个参数的含义、是否必选、默认值、取值范围和其他约束。
  4. 类型转换
    • 插件是如何在实际的存储类型和DataX的内部类型之间进行转换的。
    • 以及是否存在特殊处理。
  5. 性能报告
    • 软硬件环境,系统版本,java版本,CPU、内存等。
    • 数据特征,记录大小等。
    • 测试参数集(多组),系统参数(比如并发数),插件参数(比如batchSize)
    • 不同参数下同步速度(Rec/s, MB/s),机器负载(load, cpu)等,对数据源压力(load, cpu, mem等)。
  6. 约束限制:是否存在其他的使用限制条件。
  7. FAQ:用户经常会遇到的问题。