JAVA SDK 方式使用

JAVA SDK 方式使用

Linkis 提供了方便的JAVA和SCALA调用的接口，只需要引入linkis-computation-client的模块就可以进行使用，1.0后新增支持带Label提交的方式，下面将对兼容0.X的方式和1.0新增的方式进行介绍

1. 引入依赖模块

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

2. Java测试代码

建立Java的测试类LinkisClientTest，具体接口含义可以见注释：

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. }

运行上述的代码即可以完成任务提交/执行/日志/结果集获取等

3. Scala测试代码：

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. }