Merge pull request #201 from zhihui-ge/master

yarn session模式下可以指定applicationID提交任务

Author: yangsishu

README.md

@@ -58,7 +58,7 @@ mvn clean package -Dmaven.test.skip
 #### 1.4.1 启动命令
 
 ```
-sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack/150_flinkplugin/sqlplugin   -localSqlPluginPath D:\gitspace\flinkStreamSQL\plugins   -addjar \["udf.jar\"\] -mode yarn -flinkconf D:\flink_home\kudu150etc  -yarnconf D:\hadoop\etc\hadoopkudu -confProp \{\"time.characteristic\":\"EventTime\",\"sql.checkpoint.interval\":10000\}
+sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack/150_flinkplugin/sqlplugin   -localSqlPluginPath D:\gitspace\flinkStreamSQL\plugins   -addjar \["udf.jar\"\] -mode yarn -flinkconf D:\flink_home\kudu150etc  -yarnconf D:\hadoop\etc\hadoopkudu -confProp \{\"time.characteristic\":\"EventTime\",\"sql.checkpoint.interval\":10000\} -yarnSessionConf \{\"yid\":\"application_1564971615273_38182\"}
 ```
 
 #### 1.4.2 命令行参数选项
@@ -149,6 +149,11 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
 	* 描述：per_job 模式下指定的yarn queue
 	* 必选：否
 	* 默认值：false	
+	
+* **yarnSessionConf**
+	* 描述：yarn session 模式下指定的运行的一些参数，[可参考](https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/cli.html),目前只支持指定yid
+	* 必选：否
+	* 默认值：false	
 
 ## 2 结构
 ### 2.1 源表插件

core/src/main/java/com/dtstack/flink/sql/option/Options.java

@@ -68,6 +68,9 @@ public class Options {
     @OptionRequired(description = "yarn queue")
     private String queue = "default";
 
+    @OptionRequired(description = "yarn session configuration,such as yid")
+    private String yarnSessionConf = "{}";
+
     public String getMode() {
         return mode;
     }
@@ -171,4 +174,12 @@ public String getQueue() {
     public void setQueue(String queue) {
         this.queue = queue;
     }
+
+    public String getYarnSessionConf() {
+        return yarnSessionConf;
+    }
+
+    public void setYarnSessionConf(String yarnSessionConf) {
+        this.yarnSessionConf = yarnSessionConf;
+    }
 }

launcher/src/main/java/com/dtstack/flink/sql/launcher/ClusterClientFactory.java

@@ -51,6 +51,8 @@
 
 import com.dtstack.flink.sql.enums.ClusterMode;
 import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.util.StringHelper;
+
 import java.io.IOException;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -113,7 +115,15 @@ public static ClusterClient createYarnClient(Options launcherOptions, String mod
                     ApplicationId applicationId = null;
                     ClusterClient clusterClient = null;
                     if(mode.equals(ClusterMode.yarn.name())) {//on yarn cluster mode
-                        applicationId = getYarnClusterApplicationId(yarnClient);
+                        String yarnSessionConf = launcherOptions.getYarnSessionConf();
+                        yarnSessionConf = URLDecoder.decode(yarnSessionConf, Charsets.UTF_8.toString());
+                        Properties yarnSessionConfProperties = PluginUtil.jsonStrToObject(yarnSessionConf, Properties.class);
+                        String yid = yarnSessionConfProperties.get("yid").toString();
+                        if(StringUtils.isNotBlank(yid)){
+                            applicationId = toApplicationId(yid);
+                        }else{
+                            applicationId = getYarnClusterApplicationId(yarnClient);
+                        }
                         System.out.println("applicationId="+applicationId.toString());
 
                         AbstractYarnClusterDescriptor clusterDescriptor = new YarnClusterDescriptor(
@@ -249,4 +259,21 @@ private static org.apache.hadoop.conf.Configuration haYarnConf(org.apache.hadoop
         return yarnConf;
     }
 
+    private static ApplicationId toApplicationId(String appIdStr) {
+        Iterator<String> it = StringHelper._split(appIdStr).iterator();
+        if (!(it.next()).equals("application")) {
+            throw new IllegalArgumentException("Invalid ApplicationId prefix: " + appIdStr + ". The valid ApplicationId should start with prefix " + "application");
+        } else {
+            try {
+                return toApplicationId(it);
+            } catch (NumberFormatException e) {
+                throw new IllegalArgumentException("Invalid AppAttemptId: " + appIdStr, e);
+            }
+        }
+    }
+
+    private static ApplicationId toApplicationId(Iterator<String> it) throws NumberFormatException {
+        return ApplicationId.newInstance(Long.parseLong(it.next()), Integer.parseInt(it.next()));
+    }
+
 }