章目录

DataX的插件机制是其核心特性之一它使得DataX能够灵活地适应各种不同的数据源的数据同步。

这一机制主要基于插件开发框架该框架主要包括Reader插件、Transformer插件、Writer插件。

DataX的插件机制还采用了框架插件的架构。

框架负责连接Reader和Writer插件作为两者的数据传输通道并处理缓冲、流控、并发、数据转换等核心技术问题。

这种架构使得插件只需关心数据的读取或写入本身而同步的共性问题则由框架来处理。

此外DataX的插件机制还具有良好的扩展性和可维护性。

开发者可以根据需要开发新的Reader或Writer插件来支持新的数据源类型而无需修改DataX的核心框架代码。

这种插件化的设计使得DataX能够适应不断变化的业务需求和技术环境。

SPIService

Interface的机制。

它会在指定的插件目录中查找并加载插件然后将其注册到插件注册中心。

这样当需要使用某个插件时就可以从注册中心中获取其实例并进行相应的操作。

总的来说DataX的插件机制是一种非常灵活和可扩展的设计它使得DataX能够适应各种不同的数据源和数据存储需求同时也为开发者提供了丰富的扩展和定制化的可能性。

一、插件分类

中间结果转换例如SubstrTransformer用于字符截取

按照运行类型分

datax\plugin下分2个reader和writer目录下面以mysql为例

plugin.json内容

xxx.plugin.reader.mysqlreader.MysqlReader,description:

useScene:

DataX进程启动入口为com.alibaba.datax.core.Engineengine.entry()

public

Options();options.addOption(job,

true,

config.);options.addOption(jobid,

true,

cl.getOptionValue(jobid);RUNTIME_MODE

cl.getOptionValue(mode);Configuration

configuration

ConfigParser.parse(jobPath);}读取并解析插件配置

String

jobPath)传入job路径该方法组装解析最后返回一个Configuration对象Configuration里解析出了readerwriterhandler等插件名称提取完插件名称后会去reader寻找插件的位置。

动态加载插件

插件的加载都是通过自定义类加载器JarLoader动态加载,提供插件相关Jar隔离的加载机制。

插件的加载接口由LoadUtil类负责当要加载一个插件时需要实例化一个JarLoader然后切换thread

class

loader之后才加载插件。

这个主要由ClassLoaderSwapper实现。

JarLoader类

JarLoader

文件的合法性、有效性以及是否包含必要的插件实现类。

继承自URLClassLoader提供Jar隔离的加载机制会把传入的路径、及其子路径、以及路径中的jar文件加入到class

/***

提供Jar隔离的加载机制会把传入的路径、及其子路径、以及路径中的jar文件加入到class

public

JarLoader.class.getClassLoader());}public

JarLoader(String[]

paths.length,jar包路径不能为空.);ListString

dirs

{dirs.add(path);JarLoader.collectDirs(path,

dirs);}ListURL

{urls.addAll(doGetURLs(path));}return

urls.toArray(new

{continue;}collector.add(child.getAbsolutePath());collectDirs(child.getAbsolutePath(),

collector);}}private

{Validate.isTrue(!StringUtils.isBlank(path),

jar包路径不能为空.);File

File(path);Validate.isTrue(jarPath.exists()

jarPath.isDirectory(),jar包路径必须存在且为目录.);/*

set

pathname.getName().endsWith(.jar);}};/*

iterate

File(path).listFiles(jarFilter);ListURL

jarURLs

ArrayListURL(allJars.length);for

(int

{jarURLs.add(allJars[i].toURI().toURL());}

catch

DataXException.asDataXException(FrameworkErrorCode.PLUGIN_INIT_ERROR,系统加载jar包出错,

e);}}return

提供了加载插件的方法。

这些方法会根据配置文件中指定的插件类型和名称使用

Java

会负责创建插件的实例。

这通常涉及到调用插件类的无参构造函数并返回该实例的引用。

LoadUtil

会处理任何与实例化相关的异常以确保在出现问题时能够给出适当的错误消息。

配置传递LoadUtil

还可能负责将配置文件中针对插件的配置参数传递给插件实例。

这确保了插件能够根据用户的配置进行正确的初始化。

错误处理如果在加载、实例化或配置插件过程中发生错误LoadUtil

会负责处理这些错误。

这可能包括记录日志、抛出异常或采取其他恢复措施。

public

所有插件配置放置在pluginRegisterCenter中为区别reader、transformer和writer还能区别*

具体pluginName故使用pluginType.pluginName作为key放置在该map中*/private

static

String.format(pluginTypeNameFormat,

pluginType.toString(),pluginName);}private

static

pluginRegisterCenter.getConfiguration(generatePluginKey(pluginType,

pluginName));if

DataXException.asDataXException(FrameworkErrorCode.PLUGIN_INSTALL_ERROR,String.format(DataX不能找到插件[%s]的配置.,pluginName));}return

pluginConf;}/***

加载JobPluginreader、writer都可能要加载**

param

LoadUtil.loadPluginClass(pluginType,

pluginName,

clazz.newInstance();jobPlugin.setPluginConf(getPluginConf(pluginType,

pluginName));return

DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,String.format(DataX找到plugin[%s]的Job配置.,pluginName),

e);}}/***

加载taskPluginreader、writer都可能加载**

param

LoadUtil.loadPluginClass(pluginType,

pluginName,

clazz.newInstance();taskPlugin.setPluginConf(getPluginConf(pluginType,

pluginName));return

DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,String.format(DataX不能找plugin[%s]的Task配置.,pluginName),

e);}}/***

根据插件类型、名字和执行时taskGroupId加载对应运行器**

param

LoadUtil.loadTaskPlugin(pluginType,pluginName);switch

(pluginType)

WriterRunner(taskPlugin);default:throw

DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,String.format(插件[%s]的类型必须是[reader]或[writer]!,pluginName));}}/***

反射出具体plugin实例**

return*/SuppressWarnings(unchecked)private

static

LoadUtil.getJarLoader(pluginType,

pluginName);try

jarLoader.loadClass(pluginConf.getString(class)

catch

DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,

e);}}public

jarLoaderCenter.get(generatePluginKey(pluginType,pluginName));if

(null

(StringUtils.isBlank(pluginPath))

{throw

DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,String.format(%s插件[%s]路径非法!,pluginType,

new

String[]{pluginPath});jarLoaderCenter.put(generatePluginKey(pluginType,

jarLoader;}

ClassLoaderSwapper有一个属性storeClassLoader

用于保存着当前线程的classLoader切换之前的ClassLoader。

/***

为避免jar冲突比如hbase可能有多个版本的读写依赖jar包JobContainer和TaskGroupContainer就需要脱离当前classLoader去加载这些jar包执行完成后又退回到原来classLoader上继续执行接下来的代码*/

public

newCurrentThreadClassLoaderSwapper()

{return

保存当前classLoader并将当前线程的classLoader设置为所给classLoader**

param*

setCurrentThreadClassLoader(ClassLoader

classLoader)

Thread.currentThread().getContextClassLoader();Thread.currentThread().setContextClassLoader(classLoader);return

return*/public

restoreCurrentThreadClassLoader()

{ClassLoader

Thread.currentThread().getContextClassLoader();Thread.currentThread().setContextClassLoader(this.storeClassLoader);return

classLoader;}