datax值转换使用以及源码分析

一、DataX的数据传输基础
- 1. 通道Channel
- 2. TaskGroupContainer中的成员变量Channel分析
- 3. Channel配置
二、datax脏数据处理
- 1. 什么是脏数据？
- 2. 如何处理脏数据
- - 1）脏数据阈值控制
- 3. 类型转换（DataX的内部类型）
- 4. 插件是如何在实际的存储类型和`DataX`的内部类型之间进行转换的
三、DataX自定义transformer
- 1. 如何定义自定义的转换方法，不在datax源码中
四、TransformerUtil执行流程
- 1. 每个函数或值转换可以重复执行吗？
五、GroovyTransformer转换代码分析
六、自定义javascript函数处理转换
七、问题思考总结
- 1. datax多个字段混合转换是否支持？（一次多个输入字段或多个输出字段）
- 2. datax json配置文件中的columnIndex 从0 开始还是从1 开始？
- 3. datax 代码获取列名信息？
八、datax调试以及打印
- 1. datax debug远程调试
九、参考

一、DataX的数据传输基础

跟一般的生产者-消费者模式一样，Reader插件和Writer插件之间也是通过channel来实现数据的传输的。channel可以是内存的，也可能是持久化的，插件不必关心。插件通过RecordSender往channel写入数据，通过RecordReceiver从channel读取数据。
channel中的一条数据为一个Record的对象，Record中可以放多个Column对象，这可以简单理解为数据库中的记录和列。

传输模块transport代码目录:

1. 通道Channel

抽象类Channel，统计和限速都在这里。

一个DataX Job会切分成多个Task，每个Task会按TaskGroup进行分组，一个Task内部会有一组Reader->Channel->Writer。Channel是连接Reader和Writer的数据交换通道，所有的数据都会经由Channel进行传输。

在DataX内部对每个Channel会有严格的速度控制，分两种，一种是控制每秒同步的记录数，另外一种是每秒同步的字节数，默认的速度限制是1MB/s，可以根据具体硬件情况设置这个byte速度或者record速度，一般设置byte速度，比如：我们可以把单个Channel的速度上限配置为5MB。

当一个Job内Channel数变多后，内存的占用会显著增加，因为DataX作为数据交换通道，在内存中会缓存较多的数据。

注意：
MysqlReader进行数据抽取时，如果指定splitPk，表示用户希望使用splitPk代表的字段进行数据分片，DataX因此会启动并发任务进行数据同步，这样可以大大提供数据同步的效能，splitPk不填写，包括不提供splitPk或者splitPk值为空，DataX视作使用单通道同步该表数据，配置多channel单不配置splitPk测试不出来效果。

官方只提供了一个内存Channel的具体实现，底层其实是一个ArrayBlockingQueue

public class MemoryChannel extends Channel {
...
}

2. TaskGroupContainer中的成员变量Channel分析

对于TaskGroupContainer，每个TaskGroupContainer并发执行的个数由CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_CHANNEL决定。

            // 获取channel数目int channelNumber = this.configuration.getInt(CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_CHANNEL);

CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_CHANNEL的值如下显示为5

    private void schedule() {/*** 这里的全局speed和每个channel的速度设置为B/s*/int channelsPerTaskGroup = this.configuration.getInt(CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_CHANNEL, 5);int taskNumber = this.configuration.getList(CoreConstant.DATAX_JOB_CONTENT).size();this.needChannelNumber = Math.min(this.needChannelNumber, taskNumber);PerfTrace.getInstance().setChannelNumber(needChannelNumber);

总结：由此可以看出TaskGroupContainer默认并发的task个数是5。

3. Channel配置

DataX 内部对每个 Channel 会做限速，可以限制每秒 byte 数，也可以限制每秒 record 数。除了对每个 Channel 限速，在全局还会有一个速度限制的配置，默认是不限。
1，配置全局 Byte 限速以及单 Channel Byte 限速，Channel 个数 = 全局 Byte 限速 / 单 Channel Byte 限速。(下面示例中最终 Channel 个数为 10)

"core": {
"transport": {
"channel": {
"speed": {
"byte": 1048576
}
}
}
},
"job": {
"setting": {
"speed": {
"byte" : 10485760
}
},
...
}
}

2，配置全局 Record 限速以及单 Channel Record 限速，Channel 个数 = 全局 Record 限速 / 单 Channel Record 限速。(下面示例中最终 Channel 个数为 3)

"core": {
"transport": {
"channel": {
"speed": {
"record": 100
}
}
}
},
"job": {
"setting": {
"speed": {
"record" : 300
}
},
...
}
}

3，全局不限速，直接配置 Channel 个数。(下面示例中最终 Channel 个数为 5)

"core": {
"transport": {
"channel": {
"speed": {
"byte": 1048576
}
}
}
},
"job": {
"setting": {
"speed": {
"channel" : 5
}
},
...
}
}

第三种方式最简单直接，但是这样就缺少了全局的限速。在选择 Channel 个数时，同样需要注意，Channel 个数并不是越多越好。Channel 个数的增加，带来的是更多的 CPU 消耗以及内存消耗。如果 Channel 并发配置过高导致 JVM 内存不够用，会出现的情况是发生频繁的 Full GC，导出速度会骤降，适得其反。

可以在 DataX 的输出日志中，找到本次任务的 Channel 的数：关键字：splits to

注意：这里对一定要对tasks要有个正确的认识，根据如下代码显示，日志打印中 [1]tasks的tasks其实就是来自CoreConstant.DATAX_JOB_CONTENT（“job.content”）关键字content 数组的大小。就是说一个task对应一个content。

           List<Configuration> taskConfigs = this.configuration.getListConfiguration(CoreConstant.DATAX_JOB_CONTENT);if(LOG.isDebugEnabled()) {LOG.debug("taskGroup[{}]'s task configs[{}]", this.taskGroupId,JSON.toJSONString(taskConfigs));}int taskCountInThisTaskGroup = taskConfigs.size();LOG.info(String.format("taskGroupId=[%d] start [%d] channels for [%d] tasks.",this.taskGroupId, channelNumber, taskCountInThisTaskGroup));

二、datax脏数据处理

1. 什么是脏数据？

目前主要有三类脏数据：

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

2. 如何处理脏数据

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

1）脏数据阈值控制

job.setting.errorLimit(脏数据控制)
Job支持用户对于脏数据的自定义监控和告警，包括对脏数据最大记录数阈值（record值）或者脏数据占比阈值（percentage值），当Job传输过程出现的脏数据大于用户指定的数量/百分比，DataX Job报错退出。

3. 类型转换（DataX的内部类型）

参考： datax源码中搜关键字 “”类型转换“”
如下：来自官方文档

为了规范源端和目的端类型转换操作，保证数据不失真，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开头的数据类型转换转换方法。

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`	不支持	-

4. 插件是如何在实际的存储类型和`DataX`的内部类型之间进行转换的

关于具体每个插件 datax源码文档都有描述，具体可以查看每个插件源码目录下的md文件。
类型转换
- 插件是如何在实际的存储类型和DataX的内部类型之间进行转换的。
- 以及是否存在特殊处理。

三、DataX自定义transformer

DataX 运行时加载自定义 transformer 插件
参考URL: https://blog.csdn.net/landstream/article/details/88878172

下载源码，在根目录下找到 transformer 文件夹。

找到抽象类transformer类
参考已有的transformer类实现接口，按你的需求接收参数，用于从 job 配置文件接收命令。
在 core\src\main\java\com\alibaba\datax\core\transport\transformer 目录的 TransformerRegistry 类中注册你编写的 transformer 类。

注意：这种方式，自定义的转换类中的setTransformerName(“dx_substr”);必须以dx_开头，因此如下代码所示registTransformer在调registTransformer时的第三个参数写死时true（代表是不是datax自带方法），这个函数里面是判断如果是datax自带实现方法，那么必须以 dx_substr开头。

    public static synchronized void registTransformer(Transformer transformer) {registTransformer(transformer, null, true);}

1. 如何定义自定义的转换方法，不在datax源码中

datax其实把转换分成2类，

一类叫本地（isNative)（转换类自定义在datax中）
一类叫非本地
非本地转换容许你定义一个transformer.json，从这个json中读取有用信息。

它首先从datax json 配置 transformer下读取name，然后调
TransformerRegistry.loadTransformerFromLocalStorage(functionNames);

如下代码所示，它是判断name的值(其实是一个目录名)在路径（DATAX_HOME/local_storage/transformer/）下是否存在,如果存在调loadTransformer 加载该转换。

    public static void loadTransformerFromLocalStorage(List<String> transformers) {String[] paths = new File(CoreConstant.DATAX_STORAGE_TRANSFORMER_HOME).list();if (null == paths) {return;}for (final String each : paths) {try {if (transformers == null || transformers.contains(each)) {loadTransformer(each);}} catch (Exception e) {LOG.error(String.format("skip transformer(%s) loadTransformer has Exception(%s)", each, e.getMessage()), e);}}}

loadTransformer加载是一个目录名/transformer.json结尾的 json配置文件。
从该json配置文件中解析要加载转换类jar包，加载该jar包。

因此，自定义非本地函数，写好这个 json配置文件很重要。

四、TransformerUtil执行流程

它封装了一个处理转换的工具类：TransformerUtil，调了TransformerRegistry的两个静态方法，一个是:加载转换从本地，一个是:根据方法名获取转换实例。
TransformerRegistry.loadTransformerFromLocalStorage(functionNames);
和TransformerRegistry.getTransformer(functionName);

TransformerUtil流程
1）TransformerUtil该类,只有一个静态方法，它读取datax json配置文件的transformer key下的内容，读到一个List中。即读取所有转换配置到一个list中。
2） for遍历这个List读取name key，获取函数名List

遍历函数名list，调TransformerRegistry.loadTransformerFromLocalStorage(functionNames);加载这些类。
4）再一次遍历配置List，
获取列索引，
获取参数，
组装TransformerExecutionParas实例，该实例用于传递参数给具体转换类。
又根据transformerExecutionParas捆绑到具体使用该配置的转换实例 TransformerExecution transformerExecution = new TransformerExecution(transformerInfo, transformerExecutionParas);

public static List<TransformerExecution> buildTransformerInfo(Configuration taskConfig){List<Configuration> tfConfigs = taskConfig.getListConfiguration(CoreConstant.JOB_TRANSFORMER);...
}

TransformerExecutionParas

public class TransformerExecutionParas {/*** 以下是function参数*/private Integer columnIndex;private String[] paras;private Map<String, Object> tContext;private String code;private List<String> extraPackage;

大致总结为如下图：

1. 每个函数或值转换可以重复执行吗？

经过测试：如下dx_substr函数可以调2次，并且针对同一个字段 “columnIndex”:5。

 "transformer": [{"name": "dx_substr","parameter": {"columnIndex":5,"paras":["1","3"]}  },{"name": "dx_substr","parameter": {"columnIndex":5,"paras":["1","2"]}  }]

通过上面源码分析，和测试结果一致。for遍历用户json配置的转换（因此可以从上往下顺序执行），然后根据name字段获取函数名List，然后根据这个list加载对应类。

五、GroovyTransformer转换代码分析

Datax 自定义函数 dx_groovy
参考URL: https://www.jianshu.com/p/2b267ffda45b

datax json 配置如下,我们看到之前TransformerExecutionParas类中存储的方法参数的code、extraPackage这两个成员变量用在这里。

                "transformer": [{"name": "dx_groovy","parameter": {"code": "//groovy code//",  "extraPackage":["import somePackage1;", "import somePackage2;"]                      }  }]

Groovy转换类：

public class GroovyTransformer extends Transformer {public GroovyTransformer() {setTransformerName("dx_groovy");}...}

dx_groovy只能调用一次。不能多次调用。
groovy code中支持java.lang, java.util的包，可直接引用的对象有record，以及element下的各种 column（BoolColumn.class,BytesColumn.class,DateColumn.class,DoubleColumn.class,LongColumn.class,StringColumn.class）。.
不支持其他包，如果用户有需要用到其他包，可设置extraPackage，注意extraPackage不支持第三方jar包。
groovy code中，返回更新过的Record（比如record.setColumn(columnIndex, new StringColumn(newValue));），或者null。返回null表示过滤此行。
用户可以直接调用静态的Util方式（GroovyTransformerStaticUtil），目前GroovyTransformerStaticUtil的方法列表 (按需补充)：

如下所示通过getGroovyRule，返回类的字符串，解析这个字符串到groovyClass，获取对应的实例，执行实例的evaluatef方法（用户输入的代码封装在该方法里面）

    private void initGroovyTransformer(String code, List<String> extraPackage) {GroovyClassLoader loader = new GroovyClassLoader(GroovyTransformer.class.getClassLoader());String groovyRule = getGroovyRule(code, extraPackage);Class groovyClass;try {groovyClass = loader.parseClass(groovyRule);} catch (CompilationFailedException cfe) {throw DataXException.asDataXException(TransformerErrorCode.TRANSFORMER_GROOVY_INIT_EXCEPTION, cfe);}try {Object t = groovyClass.newInstance();if (!(t instanceof Transformer)) {throw DataXException.asDataXException(TransformerErrorCode.TRANSFORMER_GROOVY_INIT_EXCEPTION, "datax bug! contact askdatax");}this.groovyTransformer = (Transformer) t;} catch (Throwable ex) {throw DataXException.asDataXException(TransformerErrorCode.TRANSFORMER_GROOVY_INIT_EXCEPTION, ex);}}

getGroovyRule 方法：其实就是把用户定义的代码封装到RULE类下的 evaluate方法下：

六、自定义javascript函数处理转换

jdk1.6开始就提供了动态脚本语言诸如JavaScript动态的支持。
java 语言层面支持（java8 新JavaScript引擎nashorn）javascript
因此采用：使用Java自带的ScriptEngine可以说是最完美的Java动态执行代码方案。

检测用户输入的 javascript code

定义一个绑定实现 ScriptContext 接口
用来指定要传递给js的数据，以及其范围，比如引擎范围 ENGINE_SCOPE或全局 GLOBAL_SCOPE

比如：传递你要操作的列

把java 中列信息，传递给js
js处理列信息，处理列信息
js 返回每个处理后的列信息，给java，datax修改record实例
record.setColumn(columnIndex, new StringColumn(newValue));

七、问题思考总结

1. datax多个字段混合转换是否支持？（一次多个输入字段或多个输出字段）

我们分析datax 子串转换 SubstrTransformer ，如下我们看到它是，直接从paras取得第一个字符串，强转(Integer) ，因此它默认是不支持，这个SubstrTransformer 是不支持一次操作多个字段。

    @Overridepublic Record evaluate(Record record, Object... paras) {int columnIndex;int startIndex;int length;try {if (paras.length != 3) {throw new RuntimeException("dx_substr paras must be 3");}columnIndex = (Integer) paras[0];startIndex = Integer.valueOf((String) paras[1]);length = Integer.valueOf((String) paras[2]);} catch (Exception e) {throw DataXException.asDataXException(TransformerErrorCode.TRANSFORMER_ILLEGAL_PARAMETER, "paras:" + Arrays.asList(paras).toString() + " => " + e.getMessage());}Column column = record.getColumn(columnIndex);

思考：那么，如何一次处理多个字段，比如我需要把某两个字段连接起来，作为目标库中的某一个字段的值？

2. datax json配置文件中的columnIndex 从0 开始还是从1 开始？

经过测试：从 0 开始。其中 columnIndex 对应的就是reader 配置下的column配置的顺序。
查看代码如下，列用ArrayList存储，所以从0开始。

public class DefaultRecord implements Record {private static final int RECORD_AVERGAE_COLUMN_NUMBER = 16;private List<Column> columns;private int byteSize;// 首先是Record本身需要的内存private int memorySize = ClassSize.DefaultRecordHead;public DefaultRecord() {this.columns = new ArrayList<Column>(RECORD_AVERGAE_COLUMN_NUMBER);}@Overridepublic void addColumn(Column column) {columns.add(column);incrByteSize(column);}

3. datax 代码获取列名信息？

Record 代表一行记录，由列组成，列用成员变量ArrayList columns表示。

跟踪代码,查看Record 和Column 发现没有存储列名地方，也就是说record实例中没有了列名信息，定义到某个列完全使用 ArrayList数组索引。

Record 行

public class DefaultRecord implements Record {private static final int RECORD_AVERGAE_COLUMN_NUMBER = 16;private List<Column> columns;private int byteSize;// 首先是Record本身需要的内存private int memorySize = ClassSize.DefaultRecordHead;

Column 列

public abstract class Column {private Type type;private Object rawData;private int byteSize;

八、datax调试以及打印

1. datax debug远程调试

【可完全参考】参考URL:https://blog.csdn.net/gucapg/article/details/91045510

步骤主要有一下2步：

启动脚本加 -d

[root@xxx~]# python /usr/local/datax/bin/datax.py /usr/local/datax/test.json -dDataX (DATAX-OPENSOURCE-3.0), From Alibaba !
Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.local ip:  34.0.0.2
Listening for transport dt_socket at address: 9999

IDEA 配置debug配置
datax入口类：com.alibaba.datax.core.Engine
然后Edit Configure–>选择 Remote->配置debug的ip、端口以及源码工程。

先执行第一步，再执行第二步即可。

九、参考

[推荐-写的比较全面]DataX的执行流程分析
参考URL: https://www.jianshu.com/p/b10fbdee7e56
一次详细的 datax 优化
参考URL: https://xiaozhuanlan.com/topic/7860594132
Datax开发使用须知
参考URL: https://blog.csdn.net/MrZhangBaby/article/details/89638119
【Java】使用ScriptEngine动态执行代码（附Java几种动态执行代码比较）
参考URL: https://blog.csdn.net/hangvane123/article/details/84945180