Use of 1.0 SDK

Use of 1.0 SDK

Linkis provides a convenient interface for calling JAVA and SCALA. You only need to introduce the linkis-computation-client module to use it. After 1.0, the method of submitting with Label is added. The following will be compatible with 0.X and 1.0. New way to introduce

1 Introduce dependent modules

1. <dependency>
2. <groupId>com.webank.wedatasphere.linkis</groupId>
3. <artifactId>linkis-computation-client</artifactId>
4. <version>${linkis.version}</version>
5. </dependency>
6. like:
7. <dependency>
8. <groupId>com.webank.wedatasphere.linkis</groupId>
9. <artifactId>linkis-computation-client</artifactId>
10. <version>1.0.0-RC1</version>
11. </dependency>

2 Compatible with 0.X Execute method submission

2.1 Java test code

Establish the Java test class UJESClientImplTestJ, the specific interface meaning can be seen in the notes:

1. package com.webank.wedatasphere.linkis.client.test

4. import com.webank.wedatasphere.linkis.common.utils.Utils;
5. import com.webank.wedatasphere.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy;
6. import com.webank.wedatasphere.linkis.httpclient.dws.authentication.TokenAuthenticationStrategy;
7. import com.webank.wedatasphere.linkis.httpclient.dws.config.DWSClientConfig;
8. import com.webank.wedatasphere.linkis.httpclient.dws.config.DWSClientConfigBuilder;
9. import com.webank.wedatasphere.linkis.ujes.client.UJESClient;
10. import com.webank.wedatasphere.linkis.ujes.client.UJESClientImpl;
11. import com.webank.wedatasphere.linkis.ujes.client.request.JobExecuteAction;
12. import com.webank.wedatasphere.linkis.ujes.client.request.ResultSetAction;
13. import com.webank.wedatasphere.linkis.ujes.client.response.JobExecuteResult;
14. import com.webank.wedatasphere.linkis.ujes.client.response.JobInfoResult;
15. import com.webank.wedatasphere.linkis.ujes.client.response.JobProgressResult;
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. public static void main(String[] args){

30. String user = "hadoop";
31. String executeCode = "show databases;";

34. // 1. Configure DWSClientBuilder, get a DWSClientConfig through DWSClientBuilder
35. DWSClientConfig clientConfig = ((DWSClientConfigBuilder) (DWSClientConfigBuilder.newBuilder()
36. .addServerUrl("http://${ip}:${port}") //Specify ServerUrl, the address of the linkis server-side gateway, such as http://{ip}:{port}
37. .connectionTimeout(30000) //connectionTimeOut client connection timeout
38. .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES) //Whether to enable registration discovery, if enabled, the newly launched Gateway will be automatically discovered
39. .loadbalancerEnabled(true) // Whether to enable load balancing, if registration discovery is not enabled, load balancing is meaningless
40. .maxConnectionSize(5) //Specify the maximum number of connections, that is, the maximum number of concurrent
41. .retryEnabled(false).readTimeout(30000) //execution failed, whether to allow retry
42. .setAuthenticationStrategy(new StaticAuthenticationStrategy()) //AuthenticationStrategy Linkis authentication method
43. .setAuthTokenKey("${username}").setAuthTokenValue("${password}"))) //Authentication key, generally the user name; authentication value, generally the password corresponding to the user name
44. .setDWSVersion("v1").build(); //Linkis background protocol version, the current version is v1

47. // 2. Get a UJESClient through DWSClientConfig
48. UJESClient client = new UJESClientImpl(clientConfig);

51. try {
52. // 3. Start code execution
53. System.out.println("user: "+ user + ", code: [" + executeCode + "]");
54. Map<String, Object> startupMap = new HashMap<String, Object>();
55. startupMap.put("wds.linkis.yarnqueue", "default"); // Various startup parameters can be stored in startupMap, see linkis management console configuration
56. JobExecuteResult jobExecuteResult = client.execute(JobExecuteAction.builder()
57. .setCreator("linkisClient-Test") //creator, the system name of the client requesting linkis, used for system-level isolation
58. .addExecuteCode(executeCode) //ExecutionCode The code to be executed
59. .setEngineType((JobExecuteAction.EngineType) JobExecuteAction.EngineType$.MODULE$.HIVE()) // The execution engine type of the linkis that you want to request, such as Spark hive, etc.
60. .setUser(user) //User, request user; used for user-level multi-tenant isolation
61. .setStartupParams(startupMap)
62. .build());
63. System.out.println("execId: "+ jobExecuteResult.getExecID() + ", taskId:" + jobExecuteResult.taskID());

66. // 4. Get the execution status of the script
67. JobInfoResult jobInfoResult = client.getJobInfo(jobExecuteResult);
68. int sleepTimeMills = 1000;
69. while(!jobInfoResult.isCompleted()) {
70. // 5. Get the execution progress of the script
71. JobProgressResult progress = client.progress(jobExecuteResult);
72. Utils.sleepQuietly(sleepTimeMills);
73. jobInfoResult = client.getJobInfo(jobExecuteResult);
74. }

77. // 6. Get the job information of the script
78. JobInfoResult jobInfo = client.getJobInfo(jobExecuteResult);
79. // 7. Get the list of result sets (if the user submits multiple SQL at a time, multiple result sets will be generated)
80. String resultSet = jobInfo.getResultSetList(client)[0];
81. // 8. Get a specific result set through a result set information
82. Object fileContents = client.resultSet(ResultSetAction.builder().setPath(resultSet).setUser(jobExecuteResult.getUser()).build()).getFileContent();
83. System.out.println("fileContents: "+ fileContents);

85. } catch (Exception e) {
86. e.printStackTrace();
87. IOUtils.closeQuietly(client);
88. }
89. IOUtils.closeQuietly(client);
90. }
91. }

Run the above code to interact with Linkis

2.2 Scala test code

1. package com.webank.wedatasphere.linkis.client.test

4. import java.util.concurrent.TimeUnit

7. import com.webank.wedatasphere.linkis.common.utils.Utils
8. import com.webank.wedatasphere.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy
9. import com.webank.wedatasphere.linkis.httpclient.dws.config.DWSClientConfigBuilder
10. import com.webank.wedatasphere.linkis.ujes.client.UJESClient
11. import com.webank.wedatasphere.linkis.ujes.client.request.JobExecuteAction.EngineType
12. import com.webank.wedatasphere.linkis.ujes.client.request.{JobExecuteAction, ResultSetAction}
13. import org.apache.commons.io.IOUtils

16. object LinkisClientImplTest extends App {

19. var executeCode = "show databases;"
20. var user = "hadoop"

23. // 1. Configure DWSClientBuilder, get a DWSClientConfig through DWSClientBuilder
24. val clientConfig = DWSClientConfigBuilder.newBuilder()
25. .addServerUrl("http://${ip}:${port}") //Specify ServerUrl, the address of the Linkis server-side gateway, such as http://{ip}:{port}
26. .connectionTimeout(30000) //connectionTimeOut client connection timeout
27. .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES) //Whether to enable registration discovery, if enabled, the newly launched Gateway will be automatically discovered
28. .loadbalancerEnabled(true) // Whether to enable load balancing, if registration discovery is not enabled, load balancing is meaningless
29. .maxConnectionSize(5) //Specify the maximum number of connections, that is, the maximum number of concurrent
30. .retryEnabled(false).readTimeout(30000) //execution failed, whether to allow retry
31. .setAuthenticationStrategy(new StaticAuthenticationStrategy()) //AuthenticationStrategy Linkis authentication method
32. .setAuthTokenKey("${username}").setAuthTokenValue("${password}") //Authentication key, generally the user name; authentication value, generally the password corresponding to the user name
33. .setDWSVersion("v1").build() //Linkis backend protocol version, the current version is v1

36. // 2. Get a UJESClient through DWSClientConfig
37. val client = UJESClient(clientConfig)

40. try {
41. // 3. Start code execution
42. println("user: "+ user + ", code: [" + executeCode + "]")
43. val startupMap = new java.util.HashMap[String, Any]()
44. startupMap.put("wds.linkis.yarnqueue", "default") //Startup parameter configuration
45. val jobExecuteResult = client.execute(JobExecuteAction.builder()
46. .setCreator("LinkisClient-Test") //creator, requesting the system name of the Linkis client, used for system-level isolation
47. .addExecuteCode(executeCode) //ExecutionCode The code to be executed
48. .setEngineType(EngineType.SPARK) // The execution engine type of Linkis that you want to request, such as Spark hive, etc.
49. .setStartupParams(startupMap)
50. .setUser(user).build()) //User, request user; used for user-level multi-tenant isolation
51. println("execId: "+ jobExecuteResult.getExecID + ", taskId:" + jobExecuteResult.taskID)

54. // 4. Get the execution status of the script
55. var jobInfoResult = client.getJobInfo(jobExecuteResult)
56. val sleepTimeMills: Int = 1000
57. while (!jobInfoResult.isCompleted) {
58. // 5. Get the execution progress of the script
59. val progress = client.progress(jobExecuteResult)
60. val progressInfo = if (progress.getProgressInfo != null) progress.getProgressInfo.toList else List.empty
61. println("progress: "+ progress.getProgress + ", progressInfo:" + progressInfo)
62. Utils.sleepQuietly(sleepTimeMills)
63. jobInfoResult = client.getJobInfo(jobExecuteResult)
64. }
65. if (!jobInfoResult.isSucceed) {
66. println("Failed to execute job: "+ jobInfoResult.getMessage)
67. throw new Exception(jobInfoResult.getMessage)
68. }

71. // 6. Get the job information of the script
72. val jobInfo = client.getJobInfo(jobExecuteResult)
73. // 7. Get the list of result sets (if the user submits multiple SQL at a time, multiple result sets will be generated)
74. val resultSetList = jobInfoResult.getResultSetList(client)
75. println("All result set list:")
76. resultSetList.foreach(println)
77. val oneResultSet = jobInfo.getResultSetList(client).head
78. // 8. Get a specific result set through a result set information
79. val fileContents = client.resultSet(ResultSetAction.builder().setPath(oneResultSet).setUser(jobExecuteResult.getUser).build()).getFileContent
80. println("First fileContents: ")
81. println(fileContents)
82. } catch {
83. case e: Exception => {
84. e.printStackTrace()
85. }
86. }
87. IOUtils.closeQuietly(client)
88. }

3 1.0 New Submit Method

1.0 adds the client.submit method, which is used to interface with the new task execution interface of 1.0, and supports the input of Label and other parameters

3.1 Java Test Class

1. package com.webank.wedatasphere.linkis.client.test;

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

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

25. public class JavaClientTest {

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

31. String user = "hadoop";
32. String executeCode = "show tables";

35. // 1. Configure ClientBuilder and get ClientConfig
36. DWSClientConfig clientConfig = ((DWSClientConfigBuilder) (DWSClientConfigBuilder.newBuilder()
37. .addServerUrl("http://${ip}:${port}") //Specify ServerUrl, the address of the linkis server-side gateway, such as http://{ip}:{port}
38. .connectionTimeout(30000) //connectionTimeOut client connection timeout
39. .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES) //Whether to enable registration discovery, if enabled, the newly launched Gateway will be automatically discovered
40. .loadbalancerEnabled(true) // Whether to enable load balancing, if registration discovery is not enabled, load balancing is meaningless
41. .maxConnectionSize(5) //Specify the maximum number of connections, that is, the maximum number of concurrent
42. .retryEnabled(false).readTimeout(30000) //execution failed, whether to allow retry
43. .setAuthenticationStrategy(new StaticAuthenticationStrategy()) //AuthenticationStrategy Linkis authentication method
44. .setAuthTokenKey("${username}").setAuthTokenValue("${password}"))) //Authentication key, generally the user name; authentication value, generally the password corresponding to the user name
45. .setDWSVersion("v1").build(); //Linkis background protocol version, the current version is v1

48. // 2. Get a UJESClient through DWSClientConfig
49. UJESClient client = new UJESClientImpl(clientConfig);

52. try {
53. // 3. Start code execution
54. System.out.println("user: "+ user + ", code: [" + executeCode + "]");
55. Map<String, Object> startupMap = new HashMap<String, Object>();
56. // Various startup parameters can be stored in startupMap, see linkis management console configuration
57. startupMap.put("wds.linkis.yarnqueue", "q02");
58. //Specify Label
59. Map<String, Object> labels = new HashMap<String, Object>();
60. //Add the label that this execution depends on: EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel
61. labels.put(LabelKeyConstant.ENGINE_TYPE_KEY, "hive-1.2.1");
62. labels.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, "hadoop-IDE");
63. labels.put(LabelKeyConstant.ENGINE_RUN_TYPE_KEY, "hql");
64. //Specify source
65. Map<String, Object> source = new HashMap<String, Object>();
66. source.put(TaskConstant.SCRIPTPATH, "LinkisClient-test");
67. JobExecuteResult jobExecuteResult = client.submit( JobSubmitAction.builder()
68. .addExecuteCode(executeCode)
69. .setStartupParams(startupMap)
70. .setUser(user)//Job submit user
71. .addExecuteUser(user)//The actual execution user
72. .setLabels(labels)
73. .setSource(source)
74. .build()
75. );
76. System.out.println("execId: "+ jobExecuteResult.getExecID() + ", taskId:" + jobExecuteResult.taskID());

79. // 4. Get the execution status of the script
80. JobInfoResult jobInfoResult = client.getJobInfo(jobExecuteResult);
81. int sleepTimeMills = 1000;
82. while(!jobInfoResult.isCompleted()) {
83. // 5. Get the execution progress of the script
84. JobProgressResult progress = client.progress(jobExecuteResult);
85. Utils.sleepQuietly(sleepTimeMills);
86. jobInfoResult = client.getJobInfo(jobExecuteResult);
87. }

90. // 6. Get the job information of the script
91. JobInfoResult jobInfo = client.getJobInfo(jobExecuteResult);
92. // 7. Get the list of result sets (if the user submits multiple SQL at a time, multiple result sets will be generated)
93. String resultSet = jobInfo.getResultSetList(client)[0];
94. // 8. Get a specific result set through a result set information
95. Object fileContents = client.resultSet(ResultSetAction.builder().setPath(resultSet).setUser(jobExecuteResult.getUser()).build()).getFileContent();
96. System.out.println("fileContents: "+ fileContents);

99. } catch (Exception e) {
100. e.printStackTrace();
101. IOUtils.closeQuietly(client);
102. }
103. IOUtils.closeQuietly(client);
104. }
105. }

3.2 Scala Test Class

1. package com.webank.wedatasphere.linkis.client.test

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

8. import com.webank.wedatasphere.linkis.common.utils.Utils
9. import com.webank.wedatasphere.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy
10. import com.webank.wedatasphere.linkis.httpclient.dws.config.DWSClientConfigBuilder
11. import com.webank.wedatasphere.linkis.manager.label.constant.LabelKeyConstant
12. import com.webank.wedatasphere.linkis.protocol.constants.TaskConstant
13. import com.webank.wedatasphere.linkis.ujes.client.UJESClient
14. import com.webank.wedatasphere.linkis.ujes.client.request.{JobSubmitAction, ResultSetAction}
15. import org.apache.commons.io.IOUtils

20. object ScalaClientTest {

23. def main(args: Array[String]): Unit = {
24. val executeCode = "show tables"
25. val user = "hadoop"

28. // 1. Configure DWSClientBuilder, get a DWSClientConfig through DWSClientBuilder
29. val clientConfig = DWSClientConfigBuilder.newBuilder()
30. .addServerUrl("http://${ip}:${port}") //Specify ServerUrl, the address of the Linkis server-side gateway, such as http://{ip}:{port}
31. .connectionTimeout(30000) //connectionTimeOut client connection timeout
32. .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES) //Whether to enable registration discovery, if enabled, the newly launched Gateway will be automatically discovered
33. .loadbalancerEnabled(true) // Whether to enable load balancing, if registration discovery is not enabled, load balancing is meaningless
34. .maxConnectionSize(5) //Specify the maximum number of connections, that is, the maximum number of concurrent
35. .retryEnabled(false).readTimeout(30000) //execution failed, whether to allow retry
36. .setAuthenticationStrategy(new StaticAuthenticationStrategy()) //AuthenticationStrategy Linkis authentication method
37. .setAuthTokenKey("${username}").setAuthTokenValue("${password}") //Authentication key, generally the user name; authentication value, generally the password corresponding to the user name
38. .setDWSVersion("v1").build() //Linkis backend protocol version, the current version is v1

41. // 2. Get a UJESClient through DWSClientConfig
42. val client = UJESClient(clientConfig)

45. try {
46. // 3. Start code execution
47. println("user: "+ user + ", code: [" + executeCode + "]")
48. val startupMap = new java.util.HashMap[String, Any]()
49. startupMap.put("wds.linkis.yarnqueue", "q02") //Startup parameter configuration
50. //Specify Label
51. val labels: util.Map[String, Any] = new util.HashMap[String, Any]
52. //Add the label that this execution depends on, such as engineLabel
53. labels.put(LabelKeyConstant.ENGINE_TYPE_KEY, "hive-1.2.1")
54. labels.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, "hadoop-IDE")
55. labels.put(LabelKeyConstant.ENGINE_RUN_TYPE_KEY, "hql")
56. //Specify source
57. val source: util.Map[String, Any] = new util.HashMap[String, Any]
58. source.put(TaskConstant.SCRIPTPATH, "LinkisClient-test")
59. val jobExecuteResult = client.submit(JobSubmitAction.builder
60. .addExecuteCode(executeCode)
61. .setStartupParams(startupMap)
62. .setUser(user) //Job submit user
63. .addExecuteUser(user) //The actual execution user
64. .setLabels(labels)
65. .setSource(source)
66. .build) //User, requesting user; used for user-level multi-tenant isolation
67. println("execId: "+ jobExecuteResult.getExecID + ", taskId:" + jobExecuteResult.taskID)

70. // 4. Get the execution status of the script
71. var jobInfoResult = client.getJobInfo(jobExecuteResult)
72. val sleepTimeMills: Int = 1000
73. while (!jobInfoResult.isCompleted) {
74. // 5. Get the execution progress of the script
75. val progress = client.progress(jobExecuteResult)
76. val progressInfo = if (progress.getProgressInfo != null) progress.getProgressInfo.toList else List.empty
77. println("progress: "+ progress.getProgress + ", progressInfo:" + progressInfo)
78. Utils.sleepQuietly(sleepTimeMills)
79. jobInfoResult = client.getJobInfo(jobExecuteResult)
80. }
81. if (!jobInfoResult.isSucceed) {
82. println("Failed to execute job: "+ jobInfoResult.getMessage)
83. throw new Exception(jobInfoResult.getMessage)
84. }

87. // 6. Get the job information of the script
88. val jobInfo = client.getJobInfo(jobExecuteResult)
89. // 7. Get the list of result sets (if the user submits multiple SQL at a time, multiple result sets will be generated)
90. val resultSetList = jobInfoResult.getResultSetList(client)
91. println("All result set list:")
92. resultSetList.foreach(println)
93. val oneResultSet = jobInfo.getResultSetList(client).head
94. // 8. Get a specific result set through a result set information
95. val fileContents = client.resultSet(ResultSetAction.builder().setPath(oneResultSet).setUser(jobExecuteResult.getUser).build()).getFileContent
96. println("First fileContents: ")
97. println(fileContents)
98. } catch {
99. case e: Exception => {
100. e.printStackTrace()
101. }
102. }
103. IOUtils.closeQuietly(client)
104. }

107. }