JAVA SDK Manual

JAVA SDK Manual

Linkis provides a convenient interface for calling JAVA and SCALA. It can be used only by introducing the linkis-computation-client module. After 1.0, the method of submitting with Label is added. The following will introduce both ways that compatible with 0.X and newly added in 1.0.

1. Introduce dependent modules

1. <dependency>
2. <groupId>org.apache.linkis</groupId>
3. <artifactId>linkis-computation-client</artifactId>
4. <version>${linkis.version}</version>
5. </dependency>
6. Such as:
7. <dependency>
8. <groupId>org.apache.linkis</groupId>
9. <artifactId>linkis-computation-client</artifactId>
10. <version>1.0.3</version>
11. </dependency>

2. Java test code

Create the Java test class LinkisClientTest. Refer to the comments to understand the purposes of those interfaces:

1. package org.apache.linkis.client.test;

4. import org.apache.linkis.common.utils.Utils;
5. import org.apache.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy;
6. import org.apache.linkis.httpclient.dws.config.DWSClientConfig;
7. import org.apache.linkis.httpclient.dws.config.DWSClientConfigBuilder;
8. import org.apache.linkis.manager.label.constant.LabelKeyConstant;
9. import org.apache.linkis.protocol.constants.TaskConstant;
10. import org.apache.linkis.ujes.client.UJESClient;
11. import org.apache.linkis.ujes.client.UJESClientImpl;
12. import org.apache.linkis.ujes.client.request.JobSubmitAction;
13. import org.apache.linkis.ujes.client.request.JobExecuteAction;
14. import org.apache.linkis.ujes.client.request.ResultSetAction;
15. import org.apache.linkis.ujes.client.response.*;
16. import org.apache.commons.io.IOUtils;

19. import java.util.HashMap;
20. import java.util.Map;
21. import java.util.concurrent.TimeUnit;

24. public class LinkisClientTest {

27. // 1. build config: linkis gateway url
28. private static DWSClientConfig clientConfig = ((DWSClientConfigBuilder) (DWSClientConfigBuilder.newBuilder()
29. .addServerUrl("http://127.0.0.1:9001/")   //set linkis-mg-gateway url: http://{ip}:{port}
30. .connectionTimeout(30000)   //connectionTimeOut
31. .discoveryEnabled(false) //disable discovery
32. .discoveryFrequency(1, TimeUnit.MINUTES)  // discovery frequency
33. .loadbalancerEnabled(true)  // enable loadbalance
34. .maxConnectionSize(5)   // set max Connection
35. .retryEnabled(false) // set retry
36. .readTimeout(30000)  //set read timeout
37. .setAuthenticationStrategy(new StaticAuthenticationStrategy())   //AuthenticationStrategy Linkis authen suppory static and Token
38. .setAuthTokenKey("hadoop")  // set submit user
39. .setAuthTokenValue("hadoop")))  // set passwd or token (setAuthTokenValue("BML-AUTH"))
40. .setDWSVersion("v1") //linkis rest version v1
41. .build();

44. // 2. new Client(Linkis Client) by clientConfig
45. private static UJESClient client = new UJESClientImpl(clientConfig);

48. public static void main(String[] args){

51. String user = "hadoop"; // execute user
52. String executeCode = "df=spark.sql(\"show tables\")\n" +
53. "show(df)"; // code support:sql/hql/py/scala
54. try {

57. System.out.println("user : " + user + ", code : [" + executeCode + "]");
58. // 3. build job and execute
59. JobExecuteResult jobExecuteResult = toSubmit(user, executeCode);
60. //0.x:JobExecuteResult jobExecuteResult = toExecute(user, executeCode);
61. System.out.println("execId: " + jobExecuteResult.getExecID() + ", taskId: " + jobExecuteResult.taskID());
62. // 4. get job jonfo
63. JobInfoResult jobInfoResult = client.getJobInfo(jobExecuteResult);
64. int sleepTimeMills = 1000;
65. int logFromLen = 0;
66. int logSize = 100;
67. while(!jobInfoResult.isCompleted()) {
68. // 5. get progress and log
69. JobProgressResult progress = client.progress(jobExecuteResult);
70. System.out.println("progress: " + progress.getProgress());
71. JobLogResult logRes = client.log(jobExecuteResult, logFromLen, logSize);
72. logFromLen = logRes.fromLine();
73. // 0: info 1: warn 2: error 3: all
74. System.out.println(logRes.log().get(3));
75. Utils.sleepQuietly(sleepTimeMills);
76. jobInfoResult = client.getJobInfo(jobExecuteResult);
77. }

80. JobInfoResult jobInfo = client.getJobInfo(jobExecuteResult);
81. // 6. Get the result set list (if the user submits multiple SQLs at a time,
82. // multiple result sets will be generated)
83. String resultSet = jobInfo.getResultSetList(client)[0];
84. // 7. get resultContent
85. Object fileContents = client.resultSet(ResultSetAction.builder().setPath(resultSet).setUser(jobExecuteResult.getUser()).build()).getFileContent();
86. System.out.println("res: " + fileContents);
87. } catch (Exception e) {
88. e.printStackTrace();
89. IOUtils.closeQuietly(client);
90. }
91. IOUtils.closeQuietly(client);
92. }

95. /**
96. * Linkis 1.0 recommends the use of Submit method
97. */
98. private static JobExecuteResult toSubmit(String user, String code) {
99. // 1. build  params
100. // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
101. Map<String, Object> labels = new HashMap<String, Object>();
102. labels.put(LabelKeyConstant.ENGINE_TYPE_KEY, "spark-2.4.3"); // required engineType Label
103. labels.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, user + "-IDE");// required execute user and creator
104. labels.put(LabelKeyConstant.CODE_TYPE_KEY, "py"); // required codeType
105. // set start up map :engineConn start params
106. Map<String, Object> startupMap = new HashMap<String, Object>(16);
107. // Support setting engine native parameters,For example: parameters of engines such as spark/hive
108. startupMap.put("spark.executor.instances", 2);
109. // setting linkis params
110. startupMap.put("wds.linkis.rm.yarnqueue", "dws");

113. // 2. build jobSubmitAction
114. JobSubmitAction jobSubmitAction = JobSubmitAction.builder()
115. .addExecuteCode(code)
116. .setStartupParams(startupMap)
117. .setUser(user) //submit user
118. .addExecuteUser(user)  // execute user
119. .setLabels(labels)
120. .build();
121. // 3. to execute
122. return client.submit(jobSubmitAction);
123. }

126. /**
127. * Compatible with 0.X execution mode
128. */
129. private static JobExecuteResult toExecute(String user, String code) {
130. // 1. build  params
131. // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
132. Map<String, Object> labels = new HashMap<String, Object>();
133. // labels.put(LabelKeyConstant.TENANT_KEY, "fate");
134. // set start up map :engineConn start params
135. Map<String, Object> startupMap = new HashMap<String, Object>(16);
136. // Support setting engine native parameters,For example: parameters of engines such as spark/hive
137. startupMap.put("spark.executor.instances", 2);
138. // setting linkis params
139. startupMap.put("wds.linkis.rm.yarnqueue", "dws");

142. // 2. build JobExecuteAction (0.X old way of using)
143. JobExecuteAction executionAction = JobExecuteAction.builder()
144. .setCreator("IDE")  //creator, the system name of the client requesting linkis, used for system-level isolation
145. .addExecuteCode(code)   //Execution Code
146. .setEngineTypeStr("spark") // engineConn type
147. .setRunTypeStr("py") // code type
148. .setUser(user)   //execute user
149. .setStartupParams(startupMap) // start up params
150. .build();
151. executionAction.addRequestPayload(TaskConstant.LABELS, labels);
152. String body = executionAction.getRequestPayload();
153. System.out.println(body);

156. // 3. to execute
157. return client.execute(executionAction);
158. }
159. }

Run the above code to interact with Linkis

3. Scala test code

Create the Scala test class LinkisClientTest. Refer to the comments to understand the purposes of those interfaces:

1. package org.apache.linkis.client.test

4. import java.util
5. import java.util.concurrent.TimeUnit

8. import org.apache.linkis.common.utils.Utils
9. import org.apache.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy
10. import org.apache.linkis.httpclient.dws.config.DWSClientConfigBuilder
11. import org.apache.linkis.manager.label.constant.LabelKeyConstant
12. import org.apache.linkis.protocol.constants.TaskConstant
13. import org.apache.linkis.ujes.client.UJESClient
14. import org.apache.linkis.ujes.client.request._
15. import org.apache.linkis.ujes.client.response._
16. import org.apache.commons.io.IOUtils
17. import org.apache.commons.lang.StringUtils

20. object LinkisClientTest {
21. // 1. build config: linkis gateway url
22. val clientConfig = DWSClientConfigBuilder.newBuilder()
23. .addServerUrl("http://127.0.0.1:9001/")   //set linkis-mg-gateway url: http://{ip}:{port}
24. .connectionTimeout(30000)   //connectionTimeOut
25. .discoveryEnabled(false) //disable discovery
26. .discoveryFrequency(1, TimeUnit.MINUTES)  // discovery frequency
27. .loadbalancerEnabled(true)  // enable loadbalance
28. .maxConnectionSize(5)   // set max Connection
29. .retryEnabled(false) // set retry
30. .readTimeout(30000)  //set read timeout
31. .setAuthenticationStrategy(new StaticAuthenticationStrategy())   //AuthenticationStrategy Linkis authen suppory static and Token
32. .setAuthTokenKey("hadoop")  // set submit user
33. .setAuthTokenValue("hadoop")  // set passwd or token (setAuthTokenValue("BML-AUTH"))
34. .setDWSVersion("v1") //linkis rest version v1
35. .build();

38. // 2. new Client(Linkis Client) by clientConfig
39. val client = UJESClient(clientConfig)

42. def main(args: Array[String]): Unit = {
43. val user = "hadoop" // execute user
44. val executeCode = "df=spark.sql(\"show tables\")\n" +
45. "show(df)"; // code support:sql/hql/py/scala
46. try {
47. // 3. build job and execute
48. println("user : " + user + ", code : [" + executeCode + "]")
49. val jobExecuteResult = toSubmit(user, executeCode)
50. //0.X: val jobExecuteResult = toExecute(user, executeCode)
51. println("execId: " + jobExecuteResult.getExecID + ", taskId: " + jobExecuteResult.taskID)
52. // 4. get job jonfo
53. var jobInfoResult = client.getJobInfo(jobExecuteResult)
54. var logFromLen = 0
55. val logSize = 100
56. val sleepTimeMills : Int = 1000
57. while (!jobInfoResult.isCompleted) {
58. // 5. get progress and log
59. val progress = client.progress(jobExecuteResult)
60. println("progress: " + progress.getProgress)
61. val logObj = client .log(jobExecuteResult, logFromLen, logSize)
62. logFromLen = logObj.fromLine
63. val logArray = logObj.getLog
64. // 0: info 1: warn 2: error 3: all
65. if (logArray != null && logArray.size >= 4 && StringUtils.isNotEmpty(logArray.get(3))) {
66. println(s"log: ${logArray.get(3)}")
67. }
68. Utils.sleepQuietly(sleepTimeMills)
69. jobInfoResult = client.getJobInfo(jobExecuteResult)
70. }
71. if (!jobInfoResult.isSucceed) {
72. println("Failed to execute job: " + jobInfoResult.getMessage)
73. throw new Exception(jobInfoResult.getMessage)
74. }

77. // 6. Get the result set list (if the user submits multiple SQLs at a time,
78. // multiple result sets will be generated)
79. val jobInfo = client.getJobInfo(jobExecuteResult)
80. val resultSetList = jobInfoResult.getResultSetList(client)
81. println("All result set list:")
82. resultSetList.foreach(println)
83. val oneResultSet = jobInfo.getResultSetList(client).head
84. // 7. get resultContent
85. val fileContents = client.resultSet(ResultSetAction.builder().setPath(oneResultSet).setUser(jobExecuteResult.getUser).build()).getFileContent
86. println("First fileContents: ")
87. println(fileContents)
88. } catch {
89. case e: Exception => {
90. e.printStackTrace()
91. }
92. }
93. IOUtils.closeQuietly(client)
94. }

97. /**
98. * Linkis 1.0 recommends the use of Submit method
99. */
100. def toSubmit(user: String, code: String): JobExecuteResult = {
101. // 1. build  params
102. // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
103. val labels: util.Map[String, Any] = new util.HashMap[String, Any]
104. labels.put(LabelKeyConstant.ENGINE_TYPE_KEY, "spark-2.4.3"); // required engineType Label
105. labels.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, user + "-IDE");// required execute user and creator
106. labels.put(LabelKeyConstant.CODE_TYPE_KEY, "py"); // required codeType

109. val startupMap = new java.util.HashMap[String, Any]()
110. // Support setting engine native parameters,For example: parameters of engines such as spark/hive
111. startupMap.put("spark.executor.instances", 2);
112. // setting linkis params
113. startupMap.put("wds.linkis.rm.yarnqueue", "dws");
114. // 2. build jobSubmitAction
115. val jobSubmitAction = JobSubmitAction.builder
116. .addExecuteCode(code)
117. .setStartupParams(startupMap)
118. .setUser(user) //submit user
119. .addExecuteUser(user) //execute user
120. .setLabels(labels)
121. .build
122. // 3. to execute
123. client.submit(jobSubmitAction)
124. }

129. /**
130. * Compatible with 0.X execution mode
131. */
132. def toExecute(user: String, code: String): JobExecuteResult = {
133. // 1. build  params
134. // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
135. val labels = new util.HashMap[String, Any]
136. // labels.put(LabelKeyConstant.TENANT_KEY, "fate");

139. val startupMap = new java.util.HashMap[String, Any]()
140. // Support setting engine native parameters,For example: parameters of engines such as spark/hive
141. startupMap.put("spark.executor.instances", 2)
142. // setting linkis params
143. startupMap.put("wds.linkis.rm.yarnqueue", "dws")
144. // 2. build JobExecuteAction (0.X old way of using)
145. val  executionAction = JobExecuteAction.builder()
146. .setCreator("IDE")  //creator, the system name of the client requesting linkis, used for system-level isolation
147. .addExecuteCode(code)   //Execution Code
148. .setEngineTypeStr("spark") // engineConn type
149. .setRunTypeStr("py") // code type
150. .setUser(user)   //execute user
151. .setStartupParams(startupMap) // start up params
152. .build();
153. executionAction.addRequestPayload(TaskConstant.LABELS, labels);
154. // 3. to execute
155. client.execute(executionAction)
156. }

161. }