add backend for per job

Author: todd5167

core/pom.xml

@@ -108,6 +108,14 @@
             <artifactId>flink-yarn_2.11</artifactId>
             <version>${flink.version}</version>
         </dependency>
+
+        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-statebackend-rocksdb -->
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
+
     </dependencies>
 
     <build>

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

@@ -43,10 +43,6 @@ public class ConfigConstrant {
 
     public static final String FLINK_CHECKPOINT_CLEANUPMODE_KEY = "flink.checkpoint.cleanup.mode";
 
-
-
-    public static final String FLINK_CHECKPOINT_DATAURI_KEY = "flinkCheckpointDataURI";
-
     public static final String SQL_ENV_PARALLELISM = "sql.env.parallelism";
 
     public static final String SQL_MAX_ENV_PARALLELISM = "sql.max.env.parallelism";
@@ -61,6 +57,10 @@ public class ConfigConstrant {
 
     public static final String SQL_TTL_MAXTIME = "sql.ttl.max";
 
+    public static final String STATE_BACKEND_KEY = "state.backend";
+    public static final String CHECKPOINTS_DIRECTORY_KEY = "state.checkpoints.dir";
+    public static final String STATE_BACKEND_INCREMENTAL_KEY = "state.backend.incremental";
+
 
     // restart plocy
     public static final int failureRate = 3;

core/src/main/java/com/dtstack/flink/sql/enums/EStateBackend.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.enums;
+
+/**
+ *  Flink状态后端类型
+ * Date: 2019/11/15
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public enum EStateBackend {
+    MEMORY,
+    ROCKSDB,
+    FILESYSTEM;
+
+    public static EStateBackend convertFromString(String type) {
+        if(type == null) {
+            throw new RuntimeException("null StateBackend!");
+        }
+        return valueOf(type.toUpperCase());
+    }
+}
+
+

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

@@ -22,10 +22,15 @@
 
 
 import com.dtstack.flink.sql.constrant.ConfigConstrant;
+import com.dtstack.flink.sql.enums.EStateBackend;
+import org.apache.commons.lang3.BooleanUtils;
 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.contrib.streaming.state.RocksDBStateBackend;
+import org.apache.flink.runtime.state.StateBackend;
 import org.apache.flink.runtime.state.filesystem.FsStateBackend;
+import org.apache.flink.runtime.state.memory.MemoryStateBackend;
 import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.TimeCharacteristic;
 import org.apache.flink.streaming.api.environment.CheckpointConfig;
@@ -51,6 +56,7 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+
 /**
  * Reason:
  * Date: 2017/2/21
@@ -130,10 +136,13 @@ public static void openCheckpoint(StreamExecutionEnvironment env, Properties pro
             throw new RuntimeException("not support value of cleanup mode :" + cleanupModeStr);
         }
 
-        String backendPath = properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_DATAURI_KEY);
-        if(backendPath != null){
-            //set checkpoint save path on file system, 根据实际的需求设定文件路径,hdfs://, file://
-            env.setStateBackend(new FsStateBackend(backendPath));
+        String backendType = properties.getProperty(ConfigConstrant.STATE_BACKEND_KEY);
+        String checkpointDataUri = properties.getProperty(ConfigConstrant.CHECKPOINTS_DIRECTORY_KEY);
+        String backendIncremental = properties.getProperty(ConfigConstrant.STATE_BACKEND_INCREMENTAL_KEY, "true");
+
+        if(!StringUtils.isEmpty(backendType)){
+            StateBackend stateBackend = createStateBackend(backendType, checkpointDataUri, backendIncremental);
+            env.setStateBackend(stateBackend);
         }
 
     }
@@ -377,4 +386,28 @@ public static TypeInformation[] transformTypes(Class[] fieldTypes){
         return types;
     }
 
+    private static StateBackend createStateBackend(String backendType, String checkpointDataUri, String backendIncremental) throws IOException {
+        EStateBackend stateBackendType = EStateBackend.convertFromString(backendType);
+        StateBackend stateBackend = null;
+        switch (stateBackendType) {
+            case MEMORY:
+                stateBackend = new MemoryStateBackend();
+                break;
+            case FILESYSTEM:
+                checkpointDataUriEmptyCheck(checkpointDataUri, backendType);
+                stateBackend = new FsStateBackend(checkpointDataUri);
+                break;
+            case ROCKSDB:
+                checkpointDataUriEmptyCheck(checkpointDataUri, backendType);
+                stateBackend = new RocksDBStateBackend(checkpointDataUri, BooleanUtils.toBoolean(backendIncremental));
+                break;
+        }
+        return stateBackend;
+    }
+
+    private static void checkpointDataUriEmptyCheck(String checkpointDataUri, String backendType) {
+        if (StringUtils.isEmpty(checkpointDataUri)) {
+            throw new RuntimeException(backendType + " backend checkpointDataUri not null!");
+        }
+    }
 }