public static TableSink getTableSink(AbstractTargetTableInfo targetTableInfo, String localSqlRootDir, String pluginLoadMode) throws Exception {
        String pluginType = targetTableInfo.getType();
        String pluginJarDirPath = PluginUtil.getJarFileDirPath(String.format(DIR_NAME_FORMAT, pluginType), localSqlRootDir, pluginLoadMode);
        String typeNoVersion = DtStringUtil.getPluginTypeWithoutVersion(pluginType);
        String className = PluginUtil.getGenerClassName(typeNoVersion, CURR_TYPE);

        return ClassLoaderManager.newInstance(pluginJarDirPath, (cl) -> {
            Class<?> sinkClass = cl.loadClass(className);
            if(!IStreamSinkGener.class.isAssignableFrom(sinkClass)){
                throw new RuntimeException("class " + sinkClass + " not subClass of IStreamSinkGener");
            }
            IStreamSinkGener streamSinkGener = sinkClass.asSubclass(IStreamSinkGener.class).newInstance();
            Object result = streamSinkGener.genStreamSink(targetTableInfo);
            return (TableSink) result;
        });
    }
Loads jar and initializes the object according to the specified sink type
Date: 2017/3/10
Company: www.dtstack.com
@author xuchao
core/src/main/java/com/dtstack/flink/sql/sink/StreamSinkFactory.java