JAVA SDK 的使用

JAVA SDK 的使用

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

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>

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

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

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

17. import java.util.HashMap;
18. import java.util.Map;
19. import java.util.concurrent.TimeUnit;

21. public class LinkisClientTest {

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

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

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

44. String user = "hadoop"; // 用户需要和AuthTokenKey的值保持一致
45. String executeCode = "df=spark.sql(\"show tables\")\n" +
46. "show(df)"; // code support:sql/hql/py/scala
47. try {

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

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

86. private static JobExecuteResult toSubmit(String user, String code) {
87. // 1. build  params
88. // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
89. Map<String, Object> labels = new HashMap<String, Object>();
90. labels.put(LabelKeyConstant.ENGINE_TYPE_KEY, "spark-2.4.3"); // required engineType Label
91. labels.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, user + "-APPName");// required execute user and creator eg:hadoop-IDE
92. labels.put(LabelKeyConstant.CODE_TYPE_KEY, "py"); // required codeType
93. // set start up map :engineConn start params
94. Map<String, Object> startupMap = new HashMap<String, Object>(16);
95. // Support setting engine native parameters,For example: parameters of engines such as spark/hive
96. startupMap.put("spark.executor.instances", 2);
97. // setting linkis params
98. startupMap.put("wds.linkis.rm.yarnqueue", "dws");

100. // 2. build jobSubmitAction
101. JobSubmitAction jobSubmitAction = JobSubmitAction.builder()
102. .addExecuteCode(code)
103. .setStartupParams(startupMap)
104. .setUser(user) //submit user
105. .addExecuteUser(user)  // execute user
106. .setLabels(labels)
107. .build();
108. // 3. to execute
109. return client.submit(jobSubmitAction);
110. }
111. }

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

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

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

12. import java.util
13. import java.util.concurrent.TimeUnit

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

32. // 2. new Client(Linkis Client) by clientConfig
33. val client = UJESClient(clientConfig)

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

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

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

97. val startupMap = new java.util.HashMap[String, Any]()
98. // Support setting engine native parameters,For example: parameters of engines such as spark/hive
99. startupMap.put("spark.executor.instances", 2);
100. // setting linkis params
101. startupMap.put("wds.linkis.rm.yarnqueue", "default");
102. // 2. build jobSubmitAction
103. val jobSubmitAction = JobSubmitAction.builder
104. .addExecuteCode(code)
105. .setStartupParams(startupMap)
106. .setUser(user) //submit user
107. .addExecuteUser(user) //execute user
108. .setLabels(labels)
109. .build
110. // 3. to execute
111. client.submit(jobSubmitAction)
112. }
113. }