public static AbstractTableParser getSqlParser(String pluginType, String sqlRootDir, String cacheType, String pluginLoadMode) throws Exception {

        String sideOperator = ECacheType.ALL.name().equalsIgnoreCase(cacheType) ? "all" : "async";
        String pluginJarPath = PluginUtil.getSideJarFileDirPath(pluginType, sideOperator, "side", sqlRootDir, pluginLoadMode);
        String className = PluginUtil.getSqlParserClassName(pluginType, CURR_TYPE);

        return ClassLoaderManager.newInstance(pluginJarPath, (cl) -> {
            Class<?> sideParser = cl.loadClass(className);
            if (!AbstractSideTableParser.class.isAssignableFrom(sideParser)) {
                throw new RuntimeException("class " + sideParser.getName() + " not subClass of AbsSideTableParser");
            }
            return sideParser.asSubclass(AbstractTableParser.class).newInstance();
        });
    }
get specify side parser
Date: 2018/7/25
Company: www.dtstack.com

@author xuchao
core/src/main/java/com/dtstack/flink/sql/side/StreamSideFactory.java