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(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("hadoop")))  // 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"; // 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. 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. }

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

111. // 2. build jobSubmitAction
112. JobSubmitAction 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. return client.submit(jobSubmitAction);
121. }
122. }

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