public static AbstractTableParser getSqlParser(String pluginType, String sqlRootDir, String pluginLoadMode) throws Exception {
        String pluginJarPath = PluginUtil.getJarFileDirPath(String.format(DIR_NAME_FORMAT, pluginType), sqlRootDir, pluginLoadMode);
        String typeNoVersion = DtStringUtil.getPluginTypeWithoutVersion(pluginType);
        String className = PluginUtil.getSqlParserClassName(typeNoVersion, CURR_TYPE);

        return ClassLoaderManager.newInstance(pluginJarPath, (cl) -> {
            Class<?> targetParser = cl.loadClass(className);
            if(!AbstractTableParser.class.isAssignableFrom(targetParser)){
                throw new RuntimeException("class " + targetParser.getName() + " not subClass of AbsTableParser");
            }
            return targetParser.asSubclass(AbstractTableParser.class).newInstance();
        });
    }
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