Merge branch 'v1.4.1' of https://github.com/DTStack/flinkStreamSQL into v1.4.1

ysq

Author: [email protected]

README.md

@@ -87,7 +87,7 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
 * **confProp**
     * 描述：一些参数设置
     * 格式: json
-    * 必选：否
+    * 必选：是 （如无参数填写空json即可）
     * 默认值：无
     * 可选参数:
         * sql.env.parallelism: 默认并行度设置

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -21,6 +21,7 @@
 package com.dtstack.flink.sql;
 
 import com.dtstack.flink.sql.classloader.DtClassLoader;
+import com.dtstack.flink.sql.enums.ECacheType;
 import com.dtstack.flink.sql.parser.CreateFuncParser;
 import com.dtstack.flink.sql.parser.InsertSqlParser;
 import com.dtstack.flink.sql.side.SideSqlExec;
@@ -270,8 +271,9 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en
                 classPathSet.add( PluginUtil.getRemoteJarFilePath(tableInfo.getType(), TargetTableInfo.TARGET_SUFFIX, remoteSqlPluginPath));
             } else if(tableInfo instanceof SideTableInfo){
 
+                String sideOperator = ECacheType.ALL.name().equals(((SideTableInfo) tableInfo).getCacheType()) ? "all" : "async";
                 sideTableMap.put(tableInfo.getName(), (SideTableInfo) tableInfo);
-                classPathSet.add(PluginUtil.getRemoteJarFilePath(tableInfo.getType(), SideTableInfo.TARGET_SUFFIX, remoteSqlPluginPath));
+                classPathSet.add(PluginUtil.getRemoteSideJarFilePath(tableInfo.getType(), sideOperator, SideTableInfo.TARGET_SUFFIX, remoteSqlPluginPath));
             }else {
                 throw new RuntimeException("not support table type:" + tableInfo.getType());
             }

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

@@ -21,6 +21,7 @@
 package com.dtstack.flink.sql.side;
 
 import com.dtstack.flink.sql.classloader.DtClassLoader;
+import com.dtstack.flink.sql.enums.ECacheType;
 import com.dtstack.flink.sql.table.AbsSideTableParser;
 import com.dtstack.flink.sql.table.AbsTableParser;
 import com.dtstack.flink.sql.util.PluginUtil;
@@ -36,14 +37,11 @@ public class StreamSideFactory {
 
     private static final String CURR_TYPE = "side";
 
-    private static final String SIDE_DIR_TMPL = "%s%sside";
-
     public static AbsTableParser getSqlParser(String pluginType, String sqlRootDir, String cacheType) throws Exception {
 
-        cacheType = cacheType == null ? "async" : cacheType;
-        String sideDir = String.format(SIDE_DIR_TMPL, pluginType, cacheType);
+        String sideOperator = ECacheType.ALL.name().equals(cacheType) ? "all" : "async";
         ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
-        String pluginJarPath = PluginUtil.getJarFileDirPath(sideDir, sqlRootDir);
+        String pluginJarPath = PluginUtil.getSideJarFileDirPath(pluginType, sideOperator, "side", sqlRootDir);
 
         DtClassLoader dtClassLoader = (DtClassLoader) classLoader;
         PluginUtil.addPluginJar(pluginJarPath, dtClassLoader);

core/src/main/java/com/dtstack/flink/sql/util/PluginUtil.java

@@ -63,6 +63,17 @@ public static String getJarFileDirPath(String type, String sqlRootDir){
         return jarPath;
     }
 
+    public static String getSideJarFileDirPath(String pluginType, String sideOperator, String tableType, String sqlRootDir) throws MalformedURLException {
+        String dirName = sqlRootDir + SP + pluginType + sideOperator + tableType.toLowerCase();
+        File jarFile = new File(dirName);
+
+        if(!jarFile.exists()){
+            throw new RuntimeException(String.format("path %s not exists!!!", dirName));
+        }
+
+        return dirName;
+    }
+
     public static String getGenerClassName(String pluginTypeName, String type) throws IOException {
         String pluginClassName = upperCaseFirstChar(pluginTypeName) + upperCaseFirstChar(type);
         return CLASS_PRE_STR  + "." + type.toLowerCase() + "." + pluginTypeName + "." + pluginClassName;
@@ -94,12 +105,18 @@ public static Properties stringToProperties(String str) throws IOException{
         return properties;
     }
 
-    public static URL getRemoteJarFilePath(String pluginType, String tableType,String remoteSqlRootDir) throws MalformedURLException {
+    public static URL getRemoteJarFilePath(String pluginType, String tableType, String remoteSqlRootDir) throws MalformedURLException {
         String dirName = pluginType + tableType.toLowerCase();
         String jarName = String.format("%s-%s.jar", pluginType, tableType.toLowerCase());
         return new URL("file:" + remoteSqlRootDir + SP + dirName + SP + jarName);
     }
 
+    public static URL getRemoteSideJarFilePath(String pluginType, String sideOperator, String tableType, String remoteSqlRootDir) throws MalformedURLException {
+        String dirName = pluginType + sideOperator + tableType.toLowerCase();
+        String jarName = String.format("%s-%s-%s.jar", pluginType, sideOperator, tableType.toLowerCase());
+        return new URL("file:" + remoteSqlRootDir + SP + dirName + SP + jarName);
+    }
+
     public static String upperCaseFirstChar(String str){
         return str.substring(0, 1).toUpperCase() + str.substring(1);
     }

hbase/hbase-side/hbase-async-side/pom.xml

@@ -43,7 +43,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-
+                                    <exclude>org.slf4j:slf4j-log4j12</exclude>
                                 </excludes>
                             </artifactSet>
                             <filters>

launcher/job/mysqlsideSql.txt

@@ -0,0 +1,52 @@
+CREATE TABLE MyTable(
+    channel STRING,
+    pv INT,
+    xctime bigint,
+    CHARACTER_LENGTH(channel) as timeLeng,
+    WATERMARK FOR xctime AS withOffset(xctime,1000)
+ )WITH(
+    type='kafka09',
+    bootstrapServers='172.16.8.198:9092',
+    offsetReset='latest',
+    topic='nbTest1'
+ );
+CREATE TABLE MyResult(
+    channel STRING,
+    pv INT
+ )WITH(
+    type='mysql',
+    url='jdbc:mysql://172.16.8.104:3306/test?charset=utf8',
+    userName='dtstack',
+    password='abc123',
+    tableName='pv'
+ );
+
+create table sideTable(
+    channel String,
+    xccount int,
+    PRIMARY KEY(channel),
+    PERIOD FOR SYSTEM_TIME
+ )WITH(
+    type='mysql',
+    url='jdbc:mysql://172.16.8.104:3306/test?charset=utf8',
+    userName='dtstack',
+    password='abc123',
+    tableName='sidetest',
+    cache = 'LRU',
+    cacheTTLMs='10000'
+ );
+
+insert
+into
+    MyResult
+    select
+        a.channel,
+        b.xccount
+    from
+        MyTable a
+    join
+        sideTable b
+            on a.channel=b.channel
+    where
+        b.channel = 'xc'
+        and a.pv=10;