Skip to content

Commit 6e31e38

Browse files
Davidhua1996Davidhua1996
committed
Enable the streamis to support different version of Linkis
1 parent 0368a7f commit 6e31e38

File tree

5 files changed

+99
-4
lines changed

5 files changed

+99
-4
lines changed

‎streamis-jobmanager/streamis-job-launcher/streamis-job-launcher-linkis/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/launcher/linkis/conf/JobLauncherConfiguration.scala‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,10 @@ object JobLauncherConfiguration {
3535
*/
3636
val FLINK_CHECKPOINT_PATH: CommonVars[String] = CommonVars("wds.streamis.launch.flink.checkpoint.dir", "/flink/flink-checkpoints")
3737

38+
/**
39+
* Linkis release version
40+
*/
41+
val FLINK_LINKIS_RELEASE_VERSION: CommonVars[String] = CommonVars("wds.streamis.launch.flink.linkis.release.version", "")
3842
/**
3943
* Variable: savepoint path
4044
*/
@@ -44,4 +48,5 @@ object JobLauncherConfiguration {
4448
* Variable: flink app
4549
*/
4650
val VAR_FLINK_APP_NAME: CommonVars[String] = CommonVars("wds.streamis.launch.variable.flink.app.name", "flink.app.name")
51+
4752
}

‎streamis-jobmanager/streamis-job-launcher/streamis-job-launcher-linkis/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/launcher/linkis/job/manager/FlinkJobLaunchManager.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ trait FlinkJobLaunchManager extends LinkisJobLaunchManager with Logging {
4848
* @param jobState job state used to launch
4949
* @return the job id.
5050
*/
51-
override def launch(job: LaunchJob, jobState: JobState): JobClient[LinkisJobInfo] = {
51+
override def innerLaunch(job: LaunchJob, jobState: JobState): JobClient[LinkisJobInfo] = {
5252
// Transform the JobState into the params in LaunchJob
5353
Option(jobState).foreach(state => {
5454
val startUpParams = TaskUtils.getStartupMap(job.getParams)

‎streamis-jobmanager/streamis-job-launcher/streamis-job-launcher-linkis/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/launcher/linkis/job/manager/LinkisJobLaunchManager.scala‎

Lines changed: 89 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,97 @@
11
package com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.manager
22

33
import com.webank.wedatasphere.streamis.jobmanager.launcher.job.manager.JobLaunchManager
4+
import com.webank.wedatasphere.streamis.jobmanager.launcher.job.state.JobState
5+
import com.webank.wedatasphere.streamis.jobmanager.launcher.job.{JobClient, LaunchJob}
6+
import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.conf.JobLauncherConfiguration
47
import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.LinkisJobInfo
8+
import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.manager.LinkisJobLaunchManager.LINKIS_JAR_VERSION_PATTERN
9+
import org.apache.commons.io.IOUtils
10+
import org.apache.commons.lang3.StringUtils
11+
import org.apache.linkis.common.utils.{Logging, Utils}
12+
import org.apache.linkis.computation.client.LinkisJob
13+
import org.apache.linkis.protocol.utils.TaskUtils
514

6-
trait LinkisJobLaunchManager extends JobLaunchManager[LinkisJobInfo] {
15+
import java.util
16+
import scala.collection.JavaConverters._
17+
import scala.util.matching.Regex
718

19+
trait LinkisJobLaunchManager extends JobLaunchManager[LinkisJobInfo] with Logging{
20+
/**
21+
* This method is used to launch a new job.
22+
*
23+
* @param job a StreamisJob wanted to be launched.
24+
* @param jobState job state used to launch
25+
* @return the job id.
26+
*/
27+
override def launch(job: LaunchJob, jobState: JobState): JobClient[LinkisJobInfo] = {
28+
// Support different version of Linkis
29+
var linkisVersion = JobLauncherConfiguration.FLINK_LINKIS_RELEASE_VERSION.getValue
30+
if (StringUtils.isBlank(linkisVersion)) {
31+
val linkisJarPath = classOf[LinkisJob].getProtectionDomain.getCodeSource.getLocation.getPath;
32+
val lastSplit = linkisJarPath.lastIndexOf(IOUtils.DIR_SEPARATOR);
33+
if (lastSplit >= 0) {
34+
linkisVersion = linkisJarPath.substring(lastSplit + 1)
35+
}
36+
}
37+
if (StringUtils.isNotBlank(linkisVersion)) {
38+
Utils.tryAndWarn {
39+
val LINKIS_JAR_VERSION_PATTERN(version) = linkisVersion
40+
linkisVersion = version
41+
}
42+
}
43+
if (StringUtils.isNotBlank(linkisVersion)){
44+
val versionSplitter: Array[String] = linkisVersion.split("\\.")
45+
val major = Integer.valueOf(versionSplitter(0))
46+
val sub = Integer.valueOf(versionSplitter(1))
47+
val fix = Integer.valueOf(versionSplitter(2))
48+
val versionNum = major * 10000 + sub * 100 + fix
49+
info(s"Recognized the linkis release version: [${linkisVersion}, version number: [${versionNum}]")
50+
if (versionNum <= 10101){
51+
warn("Linkis version number is less than [10101], should compatible the startup params in launcher.")
52+
val startupParams = TaskUtils.getStartupMap(job.getParams)
53+
// Change the unit of memory params for linkis older version
54+
changeUnitOfMemoryToG(startupParams, "flink.taskmanager.memory")
55+
changeUnitOfMemoryToG(startupParams, "flink.jobmanager.memory")
56+
// Avoid the _FLINK_CONFIG_. prefix for linkis older version
57+
val newParams = avoidParamsPrefix(startupParams, "_FLINK_CONFIG_.")
58+
startupParams.clear();
59+
startupParams.putAll(newParams)
60+
}
61+
}
62+
innerLaunch(job, jobState)
63+
}
864

65+
private def changeUnitOfMemoryToG(params: util.Map[String, Any], name: String): Unit = {
66+
params.get(name) match {
67+
case memory: String =>
68+
var actualMem = Integer.valueOf(memory) / 1024
69+
actualMem = if (actualMem <= 0) 1 else actualMem
70+
info(s"Change the unit of startup param: [${name}], value [${memory}] => [${actualMem}]")
71+
params.put(name, actualMem)
72+
case _ => // Ignores
73+
}
74+
}
75+
76+
/**
77+
* Avoid params prefix
78+
* @param params params
79+
* @param prefix prefix
80+
*/
81+
private def avoidParamsPrefix(params: util.Map[String, Any], prefix: String): util.Map[String, Any] = {
82+
params.asScala.map{
83+
case (key, value) =>
84+
if (key.startsWith(prefix)){
85+
info(s"Avoid the prefix of startup param: [${key}] => [${key.substring(prefix.length)}]")
86+
(key.substring(prefix.length), value)
87+
} else {
88+
(key, value)
89+
}
90+
}.toMap.asJava
91+
}
92+
def innerLaunch(job: LaunchJob, jobState: JobState): JobClient[LinkisJobInfo]
93+
}
94+
95+
object LinkisJobLaunchManager{
96+
val LINKIS_JAR_VERSION_PATTERN: Regex = "^[\\s\\S]*([\\d]+\\.[\\d]+\\.[\\d]+)[\\s\\S]*$".r
997
}

‎streamis-jobmanager/streamis-job-launcher/streamis-job-launcher-linkis/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/launcher/linkis/job/manager/SimpleFlinkJobLaunchManager.scala‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -130,5 +130,4 @@ class SimpleFlinkJobLaunchManager extends FlinkJobLaunchManager {
130130
object SimpleFlinkJobLaunchManager{
131131

132132
val INSTANCE_NAME = "simpleFlink";
133-
134133
}

‎streamis-server/src/main/resources/linkis.properties‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,4 +59,7 @@ wds.linkis.server.mybatis.BasePackage=com.webank.wedatasphere.streamis.datasourc
5959
com.webank.wedatasphere.streamis.projectmanager.dao
6060

6161
# Make sure that can fetch the application info finally
62-
wds.streamis.application.info.fetch.max=20
62+
wds.streamis.application.info.fetch.max=20
63+
64+
# To use the complete features of streamis in linkis 1.1.2
65+
#wds.streamis.launch.flink.linkis.release.version=1.1.2

0 commit comments

Comments
 (0)