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(true) // 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("123456")))  // set passwd or token (setAuthTokenValue("test"))
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"; // 用户需要和AuthTokenKey的值保持一致
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. System.out.println("execId: " + jobExecuteResult.getExecID() + ", taskId: " + jobExecuteResult.taskID());
61. // 4. get job jonfo
62. JobInfoResult jobInfoResult = client.getJobInfo(jobExecuteResult);
63. int sleepTimeMills = 1000;
64. int logFromLen = 0;
65. int logSize = 100;
66. while (!jobInfoResult.isCompleted()) {
67. // 5. get progress and log
68. JobProgressResult progress = client.progress(jobExecuteResult);
69. System.out.println("progress: " + progress.getProgress());
70. JobLogResult logRes = client.log(jobExecuteResult, logFromLen, logSize);
71. logFromLen = logRes.fromLine();
72. // 0: info 1: warn 2: error 3: all
73. System.out.println(logRes.log().get(3));
74. Utils.sleepQuietly(sleepTimeMills);
75. jobInfoResult = client.getJobInfo(jobExecuteResult);
76. }

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

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

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

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

3. Scala测试代码：

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

4. import org.apache.commons.io.IOUtils
5. import org.apache.commons.lang3.StringUtils
6. import org.apache.linkis.common.utils.Utils
7. import org.apache.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy
8. import org.apache.linkis.httpclient.dws.config.DWSClientConfigBuilder
9. import org.apache.linkis.manager.label.constant.LabelKeyConstant
10. import org.apache.linkis.ujes.client.request._
11. import org.apache.linkis.ujes.client.response._

14. import java.util
15. import java.util.concurrent.TimeUnit

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

36. // 2. new Client(Linkis Client) by clientConfig
37. val client = UJESClient(clientConfig)

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

75. // 6. Get the result set list (if the user submits multiple SQLs at a time,
76. // multiple result sets will be generated)
77. val jobInfo = client.getJobInfo(jobExecuteResult)
78. val resultSetList = jobInfoResult.getResultSetList(client)
79. println("All result set list:")
80. resultSetList.foreach(println)
81. val oneResultSet = jobInfo.getResultSetList(client).head
82. // 7. get resultContent
83. val resultSetResult: ResultSetResult = client.resultSet(ResultSetAction.builder.setPath(oneResultSet).setUser(jobExecuteResult.getUser).build)
84. println("metadata: " + resultSetResult.getMetadata) // column name type
85. println("res: " + resultSetResult.getFileContent) //row data
86. } catch {
87. case e: Exception => {
88. e.printStackTrace() //please use log
89. }
90. }
91. IOUtils.closeQuietly(client)
92. }

97. def toSubmit(user: String, code: String): JobExecuteResult = {
98. // 1. build  params
99. // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
100. val labels: util.Map[String, Any] = new util.HashMap[String, Any]
101. labels.put(LabelKeyConstant.ENGINE_TYPE_KEY, "spark-2.4.3"); // required engineType Label
102. labels.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, user + "-APPName"); // 请求的用户和应用名，两个参数都不能少，其中APPName不能带有"-"建议替换为"_"
103. labels.put(LabelKeyConstant.CODE_TYPE_KEY, "py"); // 指定脚本类型

106. val startupMap = new java.util.HashMap[String, Any]()
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", "default");
111. // 2. build jobSubmitAction
112. val jobSubmitAction = JobSubmitAction.builder
113. .addExecuteCode(code)
114. .setStartupParams(startupMap)
115. .setUser(user) //submit user
116. .addExecuteUser(user) //execute user
117. .setLabels(labels)
118. .build
119. // 3. to execute
120. client.submit(jobSubmitAction)
121. }
122. }