DataX插件二次开发指南

标签：插件 Task 框架 -- Job DataX 二次开发

一、 `DataX`为什么要使用插件机制？

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

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

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

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

二、插件视角看框架

逻辑执行模型

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

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两部分逻辑。

物理执行模型

框架为插件提供物理上的执行能力（线程）。DataX框架有三种运行模式：

Standalone: 单进程运行，没有外部依赖。
Local: 单进程运行，统计信息、错误信息汇报到集中存储。
Distrubuted: 分布式多进程运行，依赖DataX Service服务。

当然，上述三种模式对插件的编写而言没有什么区别，你只需要避开一些小错误，插件就能够在单机/分布式之间无缝切换了。
当JobContainer和TaskGroupContainer运行在同一个进程内时，就是单机模式（Standalone和Local）；当它们分布在不同的进程中执行时，就是分布式（Distributed）模式。

编程接口

Job和Task的逻辑是怎么对应到具体的代码中的？

首先，插件的入口类必须扩展Reader或Writer抽象类，并且实现分别实现Job和Task两个内部抽象类，Job和Task的实现必须是 内部类 的形式，原因见 加载原理 一节。以Reader为例：

public class SomeReader 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: 全局准备工作，比如mysqlwriter在写入新数据之前执行一个truncate table的操作、读取Hive数据之前，完成Kerberos认证等。
split: 拆分Task。参数adviceNumber框架建议的拆分数，一般是运行时所配置的并发度。值返回的是Task的配置列表。
post: 全局的后置工作，比如mysqlwriter同步完影子表后的rename操作。
destroy: Job对象自身的销毁工作。

Task接口功能如下：

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

需要注意的是：

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

框架按照如下的顺序执行Job和Task的接口：

DataXReaderWriter (2)

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

插件定义

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

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

{
    "name": "mysqlwriter",
    "class": "com.alibaba.datax.plugin.writer.mysqlwriter.MysqlWriter",
    "description": "Use Jdbc connect to database, execute insert sql.",
    "developer": "alibaba"
}

name: 插件名称，大小写敏感。框架根据用户在配置文件中指定的名称来搜寻插件。 十分重要 。
class: 入口类的全限定名称，框架通过反射插件入口类的实例。十分重要 。
description: 描述信息。
developer: 开发人员。

打包发布

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
    |   `-- mysqlreader
    |       |-- libs
    |       |   `-- mysql-reader-plugin-dependencies.jar
    |       |-- mysqlreader-0.0.1-SNAPSHOT.jar
    |       `-- plugin.json
    `-- writer
        |-- mysqlwriter
        |   |-- libs
        |   |   `-- mysql-writer-plugin-dependencies.jar
        |   |-- mysqlwriter-0.0.1-SNAPSHOT.jar
        |   `-- plugin.json
        |-- oceanbasewriter
        `-- odpswriter

${DATAX_HOME}/bin: 可执行程序目录。
${DATAX_HOME}/conf: 框架配置目录。
${DATAX_HOME}/lib: 框架依赖库目录。
${DATAX_HOME}/plugin: 插件目录。

插件目录分为reader和writer子目录，读写插件分别存放。插件目录规范如下：

${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": "",
            "accessId": "",
            "column": [""],
            "isCompress": "",
            "odpsServer": "",
            "partition": [
              ""
            ],
            "project": "",
            "table": "",
            "tunnelServer": ""
          }
        },
        "writer": {
          "name": "oraclewriter",
          "parameter": {
            "username": "",
            "password": "",
            "column": ["*"],
            "connection": [
              {
                "jdbcUrl": "",
                "table": [
                  ""
                ]
              }
            ]
          }
        }
      }
    ]
  }
}

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

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

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

如何设计配置参数

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

任务配置中reader和writer下parameter部分是插件的配置参数，插件的配置参数应当遵循以下原则：

驼峰命名：所有配置项采用驼峰命名法，首字母小写，单词首字母大写。
正交原则：配置项必须正交，功能没有重复，没有潜规则。
富类型：合理使用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"
      ]
    }
  ]
}

如何使用`Configuration`类

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

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

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

比如操作如下json：

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

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

x：4
a.b.c：2
a.b.c.d：null
a.b.f[0]：1
a.b.f[2].g：true

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

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

插件数据传输

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

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()方法会阻塞直到有数据。如果传输已经结束，会返回null，Writer插件可以据此判断是否结束startWrite方法。

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

类型转换

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

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

对应地，有DateColumn、LongColumn、DoubleColumn、BytesColumn、StringColumn和BoolColumn六种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	不支持	`true`为`1L`，否则`0L`		`true`为`1.0`，否则`0.0`	不支持	-

脏数据处理

什么是脏数据？

目前主要有三类脏数据：

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

如何处理脏数据

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

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

加载原理

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

三、插件介绍文档

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

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

标签：插件,Task,框架,--,Job,DataX,二次开发
From： https://www.cnblogs.com/wxm2270/p/17097444.html

一、 `DataX`为什么要使用插件机制？

二、插件视角看框架

逻辑执行模型

物理执行模型

编程接口

插件定义

打包发布

配置文件

如何设计配置参数

如何使用`Configuration`类

插件数据传输

类型转换

脏数据处理

什么是脏数据？

如何处理脏数据

加载原理

三、插件介绍文档

相关文章

赞助商

阅读排行

DataX插件二次开发指南

一、 DataX为什么要使用插件机制？

二、插件视角看框架

逻辑执行模型

物理执行模型

编程接口

插件定义

打包发布

配置文件

如何设计配置参数

如何使用Configuration类

插件数据传输

类型转换

脏数据处理

什么是脏数据？

如何处理脏数据

加载原理

三、插件介绍文档

相关文章

赞助商

阅读排行

一、 `DataX`为什么要使用插件机制？

如何使用`Configuration`类