state setIdleStateRetentionTime

Author: simenliuxing

README.md

@@ -18,6 +18,7 @@
   * 8.支持cep处理
   * 9.支持udaf
   * 10.支持谓词下移
+  * 11.支持状态的ttl
 
  ## BUG修复：
   * 1.修复不能解析sql中orderby,union语法。
@@ -109,6 +110,8 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
     * 必选：是 （如无参数填写空json即可）
     * 默认值：无
     * 可选参数:
+        * sql.ttl.min: 最小过期时间,大于0的整数,如1d、1h(d\D:天,h\H:小时,m\M:分钟,s\s:秒)
+        * sql.ttl.max: 最大过期时间,大于0的整数,如2d、2h(d\D:天,h\H:小时,m\M:分钟,s\s:秒),需同时设置最小时间,且比最小时间大5分钟
         * sql.env.parallelism: 默认并行度设置
         * sql.max.env.parallelism: 最大并行度设置
         * time.characteristic: 可选值[ProcessingTime|IngestionTime|EventTime]

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

@@ -66,6 +66,7 @@
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamContextEnvironment;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.StreamQueryConfig;
 import org.apache.flink.table.api.Table;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 import org.apache.flink.table.sinks.TableSink;
@@ -126,7 +127,7 @@ public static void main(String[] args) throws Exception {
         confProp = URLDecoder.decode(confProp, Charsets.UTF_8.toString());
         Properties confProperties = PluginUtil.jsonStrToObject(confProp, Properties.class);
         StreamExecutionEnvironment env = getStreamExeEnv(confProperties, deployMode);
-        StreamTableEnvironment tableEnv = StreamTableEnvironment.getTableEnvironment(env);
+        StreamTableEnvironment tableEnv = getStreamTableEnv(confProperties, env);
 
         List<URL> jarURList = Lists.newArrayList();
         SqlTree sqlTree = SqlParser.parseSql(sql);
@@ -348,4 +349,17 @@ private static StreamExecutionEnvironment getStreamExeEnv(Properties confPropert
         return env;
     }
 
+    /**
+     * 获取StreamTableEnvironment并设置相关属性
+     *
+     * @param confProperties
+     * @param env
+     * @return
+     */
+    private static StreamTableEnvironment getStreamTableEnv(Properties confProperties, StreamExecutionEnvironment env) {
+        confProperties = PropertiesUtils.propertiesTrim(confProperties);
+        StreamTableEnvironment tableEnv = StreamTableEnvironment.getTableEnvironment(env);
+        FlinkUtil.setTableEnvTTL(confProperties, tableEnv);
+        return tableEnv;
+    }
 }

core/src/main/java/com/dtstack/flink/sql/constrant/ConfigConstrant.java

@@ -57,6 +57,11 @@ public class ConfigConstrant {
 
     public static final String FLINK_TIME_CHARACTERISTIC_KEY = "time.characteristic";
 
+    public static final String SQL_TTL_MINTIME = "sql.ttl.min";
+
+    public static final String SQL_TTL_MAXTIME = "sql.ttl.max";
+
+
     // restart plocy
     public static final int failureRate = 3;
 

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

@@ -21,15 +21,16 @@
 package com.dtstack.flink.sql.util;
 
 
-import com.dtstack.flink.sql.classloader.ClassLoaderManager;
 import com.dtstack.flink.sql.constrant.ConfigConstrant;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.api.common.time.Time;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.runtime.state.filesystem.FsStateBackend;
 import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.TimeCharacteristic;
 import org.apache.flink.streaming.api.environment.CheckpointConfig;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.StreamQueryConfig;
 import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.api.java.BatchTableEnvironment;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
@@ -47,6 +48,8 @@
 import java.net.URLClassLoader;
 import java.util.List;
 import java.util.Properties;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 /**
  * Reason:
@@ -59,6 +62,9 @@ public class FlinkUtil {
 
     private static final Logger logger = LoggerFactory.getLogger(FlinkUtil.class);
 
+    private static final String TTL_PATTERN_STR = "^+?([1-9][0-9]*)([dDhHmMsS])$";
+    private static final Pattern TTL_PATTERN = Pattern.compile(TTL_PATTERN_STR);
+
     /**
      * 开启checkpoint
      * @param env
@@ -257,6 +263,68 @@ public static int getEnvParallelism(Properties properties){
         return StringUtils.isNotBlank(parallelismStr)?Integer.parseInt(parallelismStr):1;
     }
 
+    /**
+     * 设置ttl
+     * @param properties
+     * @param tableEnv
+     * @return
+     */
+    public static void setTableEnvTTL(Properties properties, StreamTableEnvironment tableEnv) {
+        String ttlMintimeStr = properties.getProperty(ConfigConstrant.SQL_TTL_MINTIME);
+        String ttlMaxtimeStr = properties.getProperty(ConfigConstrant.SQL_TTL_MAXTIME);
+        if (StringUtils.isNotEmpty(ttlMintimeStr) || StringUtils.isNotEmpty(ttlMaxtimeStr)) {
+            verityTtl(ttlMintimeStr, ttlMaxtimeStr);
+            Matcher ttlMintimeStrMatcher = TTL_PATTERN.matcher(ttlMintimeStr);
+            Matcher ttlMaxtimeStrMatcher = TTL_PATTERN.matcher(ttlMaxtimeStr);
+
+            Long ttlMintime = 0L;
+            Long ttlMaxtime = 0L;
+            if (ttlMintimeStrMatcher.find()) {
+                ttlMintime = getTtlTime(Integer.parseInt(ttlMintimeStrMatcher.group(1)), ttlMintimeStrMatcher.group(2));
+            }
+            if (ttlMaxtimeStrMatcher.find()) {
+                ttlMaxtime = getTtlTime(Integer.parseInt(ttlMaxtimeStrMatcher.group(1)), ttlMaxtimeStrMatcher.group(2));
+            }
+            if (0L != ttlMintime && 0L != ttlMaxtime) {
+                StreamQueryConfig qConfig = tableEnv.queryConfig();
+                qConfig.withIdleStateRetentionTime(Time.milliseconds(ttlMintime), Time.milliseconds(ttlMaxtime));
+            }
+        }
+    }
+
+    /**
+     * ttl 校验
+     * @param ttlMintimeStr 最小时间
+     * @param ttlMaxtimeStr 最大时间
+     */
+    private static void verityTtl(String ttlMintimeStr, String ttlMaxtimeStr) {
+        if (null == ttlMintimeStr
+                || null == ttlMaxtimeStr
+                || !TTL_PATTERN.matcher(ttlMintimeStr).find()
+                || !TTL_PATTERN.matcher(ttlMaxtimeStr).find()) {
+            throw new RuntimeException("sql.ttl.min 、sql.ttl.max must be set at the same time . example sql.ttl.min=1h,sql.ttl.max=2h");
+        }
+    }
+
+    /**
+     * 不同单位时间到毫秒的转换
+     * @param timeNumber 时间值，如：30
+     * @param timeUnit 单位，d:天，h:小时，m:分，s:秒
+     * @return
+     */
+    private static Long getTtlTime(Integer timeNumber,String timeUnit) {
+        if (timeUnit.equalsIgnoreCase("d")) {
+            return timeNumber * 1000l * 60 * 60 * 24;
+        } else if (timeUnit.equalsIgnoreCase("h")) {
+            return timeNumber * 1000l * 60 * 60;
+        } else if (timeUnit.equalsIgnoreCase("m")) {
+            return timeNumber * 1000l * 60;
+        } else if (timeUnit.equalsIgnoreCase("s")) {
+            return timeNumber * 1000l;
+        } else {
+            throw new RuntimeException("not support "+timeNumber+timeUnit);
+        }
+    }
 
     /**
      * 最大并发度