1.0 SDK的使用

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

1 引入依赖模块

1. <dependency>
2. <groupId>com.webank.wedatasphere.linkis</groupId>
3. <artifactId>linkis-computation-client</artifactId>
4. <version>${linkis.version}</version>
5. </dependency>
6. 如：
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 兼容0.X的Execute方法提交

2.1 Java测试代码

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

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. 配置DWSClientBuilder，通过DWSClientBuilder获取一个DWSClientConfig
35. DWSClientConfig clientConfig = ((DWSClientConfigBuilder) (DWSClientConfigBuilder.newBuilder()
36. .addServerUrl("http://${ip}:${port}")  //指定ServerUrl，linkis服务器端网关的地址,如http://{ip}:{port}
37. .connectionTimeout(30000)   //connectionTimeOut 客户端连接超时时间
38. .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES)  //是否启用注册发现，如果启用，会自动发现新启动的Gateway
39. .loadbalancerEnabled(true)  // 是否启用负载均衡，如果不启用注册发现，则负载均衡没有意义
40. .maxConnectionSize(5)   //指定最大连接数，即最大并发数
41. .retryEnabled(false).readTimeout(30000)   //执行失败，是否允许重试
42. .setAuthenticationStrategy(new StaticAuthenticationStrategy())   //AuthenticationStrategy Linkis认证方式
43. .setAuthTokenKey("${username}").setAuthTokenValue("${password}")))  //认证key，一般为用户名;  认证value，一般为用户名对应的密码
44. .setDWSVersion("v1").build();  //linkis后台协议的版本，当前版本为v1

47. // 2. 通过DWSClientConfig获取一个UJESClient
48. UJESClient client = new UJESClientImpl(clientConfig);

51. try {
52. // 3. 开始执行代码
53. System.out.println("user : " + user + ", code : [" + executeCode + "]");
54. Map<String, Object> startupMap = new HashMap<String, Object>();
55. startupMap.put("wds.linkis.yarnqueue", "default"); // 在startupMap可以存放多种启动参数，参见linkis管理台配置
56. JobExecuteResult jobExecuteResult = client.execute(JobExecuteAction.builder()
57. .setCreator("linkisClient-Test")  //creator，请求linkis的客户端的系统名，用于做系统级隔离
58. .addExecuteCode(executeCode)   //ExecutionCode 请求执行的代码
59. .setEngineType((JobExecuteAction.EngineType) JobExecuteAction.EngineType$.MODULE$.HIVE()) // 希望请求的linkis的执行引擎类型，如Spark hive等
60. .setUser(user)   //User，请求用户；用于做用户级多租户隔离
61. .setStartupParams(startupMap)
62. .build());
63. System.out.println("execId: " + jobExecuteResult.getExecID() + ", taskId: " + jobExecuteResult.taskID());

66. // 4. 获取脚本的执行状态
67. JobInfoResult jobInfoResult = client.getJobInfo(jobExecuteResult);
68. int sleepTimeMills = 1000;
69. while(!jobInfoResult.isCompleted()) {
70. // 5. 获取脚本的执行进度
71. JobProgressResult progress = client.progress(jobExecuteResult);
72. Utils.sleepQuietly(sleepTimeMills);
73. jobInfoResult = client.getJobInfo(jobExecuteResult);
74. }

77. // 6. 获取脚本的Job信息
78. JobInfoResult jobInfo = client.getJobInfo(jobExecuteResult);
79. // 7. 获取结果集列表（如果用户一次提交多个SQL，会产生多个结果集）
80. String resultSet = jobInfo.getResultSetList(client)[0];
81. // 8. 通过一个结果集信息，获取具体的结果集
82. Object fileContents = client.resultSet(ResultSetAction.builder().setPath(resultSet).setUser(jobExecuteResult.getUser()).build()).getFileContent();
83. System.out.println("fileContents: " + fileContents);

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

运行上述的代码即可以和Linkis进行交互

2.2 Scala测试代码

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. 配置DWSClientBuilder，通过DWSClientBuilder获取一个DWSClientConfig
24. val clientConfig = DWSClientConfigBuilder.newBuilder()
25. .addServerUrl("http://${ip}:${port}")  //指定ServerUrl，Linkis服务器端网关的地址,如http://{ip}:{port}
26. .connectionTimeout(30000)  //connectionTimeOut 客户端连接超时时间
27. .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES)  //是否启用注册发现，如果启用，会自动发现新启动的Gateway
28. .loadbalancerEnabled(true)  // 是否启用负载均衡，如果不启用注册发现，则负载均衡没有意义
29. .maxConnectionSize(5)   //指定最大连接数，即最大并发数
30. .retryEnabled(false).readTimeout(30000)   //执行失败，是否允许重试
31. .setAuthenticationStrategy(new StaticAuthenticationStrategy())  //AuthenticationStrategy Linkis认证方式
32. .setAuthTokenKey("${username}").setAuthTokenValue("${password}")  //认证key，一般为用户名;  认证value，一般为用户名对应的密码
33. .setDWSVersion("v1").build()  //Linkis后台协议的版本，当前版本为v1

36. // 2. 通过DWSClientConfig获取一个UJESClient
37. val client = UJESClient(clientConfig)

39. try {
40. // 3. 开始执行代码
41. println("user : " + user + ", code : [" + executeCode + "]")
42. val startupMap = new java.util.HashMap[String, Any]()
43. startupMap.put("wds.linkis.yarnqueue", "default") //启动参数配置
44. val jobExecuteResult = client.execute(JobExecuteAction.builder()
45. .setCreator("LinkisClient-Test")  //creator，请求Linkis的客户端的系统名，用于做系统级隔离
46. .addExecuteCode(executeCode)   //ExecutionCode 请求执行的代码
47. .setEngineType(EngineType.SPARK) // 希望请求的Linkis的执行引擎类型，如Spark hive等
48. .setStartupParams(startupMap)
49. .setUser(user).build())  //User，请求用户；用于做用户级多租户隔离
50. println("execId: " + jobExecuteResult.getExecID + ", taskId: " + jobExecuteResult.taskID)

52. // 4. 获取脚本的执行状态
53. var jobInfoResult = client.getJobInfo(jobExecuteResult)
54. val sleepTimeMills : Int = 1000
55. while (!jobInfoResult.isCompleted) {
56. // 5. 获取脚本的执行进度
57. val progress = client.progress(jobExecuteResult)
58. val progressInfo = if (progress.getProgressInfo != null) progress.getProgressInfo.toList else List.empty
59. println("progress: " + progress.getProgress + ", progressInfo: " + progressInfo)
60. Utils.sleepQuietly(sleepTimeMills)
61. jobInfoResult = client.getJobInfo(jobExecuteResult)
62. }
63. if (!jobInfoResult.isSucceed) {
64. println("Failed to execute job: " + jobInfoResult.getMessage)
65. throw new Exception(jobInfoResult.getMessage)
66. }

69. // 6. 获取脚本的Job信息
70. val jobInfo = client.getJobInfo(jobExecuteResult)
71. // 7. 获取结果集列表（如果用户一次提交多个SQL，会产生多个结果集）
72. val resultSetList = jobInfoResult.getResultSetList(client)
73. println("All result set list:")
74. resultSetList.foreach(println)
75. val oneResultSet = jobInfo.getResultSetList(client).head
76. // 8. 通过一个结果集信息，获取具体的结果集
77. val fileContents = client.resultSet(ResultSetAction.builder().setPath(oneResultSet).setUser(jobExecuteResult.getUser).build()).getFileContent
78. println("First fileContents: ")
79. println(fileContents)
80. } catch {
81. case e: Exception => {
82. e.printStackTrace()
83. }
84. }
85. IOUtils.closeQuietly(client)
86. }

3 1.0新增的Submit方式

1.0 新增了client.submit方法，用于对接1.0新的任务执行接口，支持传入Label等参数

3.1 Java测试类

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. 配置ClientBuilder，获取ClientConfig
36. DWSClientConfig clientConfig = ((DWSClientConfigBuilder) (DWSClientConfigBuilder.newBuilder()
37. .addServerUrl("http://${ip}:${port}")  //指定ServerUrl，linkis服务器端网关的地址,如http://{ip}:{port}
38. .connectionTimeout(30000)   //connectionTimeOut 客户端连接超时时间
39. .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES)  //是否启用注册发现，如果启用，会自动发现新启动的Gateway
40. .loadbalancerEnabled(true)  // 是否启用负载均衡，如果不启用注册发现，则负载均衡没有意义
41. .maxConnectionSize(5)   //指定最大连接数，即最大并发数
42. .retryEnabled(false).readTimeout(30000)   //执行失败，是否允许重试
43. .setAuthenticationStrategy(new StaticAuthenticationStrategy())   //AuthenticationStrategy Linkis认证方式
44. .setAuthTokenKey("${username}").setAuthTokenValue("${password}")))  //认证key，一般为用户名;  认证value，一般为用户名对应的密码
45. .setDWSVersion("v1").build();  //linkis后台协议的版本，当前版本为v1

48. // 2. 通过DWSClientConfig获取一个UJESClient
49. UJESClient client = new UJESClientImpl(clientConfig);

52. try {
53. // 3. 开始执行代码
54. System.out.println("user : " + user + ", code : [" + executeCode + "]");
55. Map<String, Object> startupMap = new HashMap<String, Object>();
56. // 在startupMap可以存放多种启动参数，参见linkis管理台配置
57. startupMap.put("wds.linkis.yarnqueue", "q02");
58. //指定Label
59. Map<String, Object> labels = new HashMap<String, Object>();
60. //添加本次执行所依赖的的标签: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. //指定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提交用户
71. .addExecuteUser(user)//实际执行用户
72. .setLabels(labels)
73. .setSource(source)
74. .build()
75. );
76. System.out.println("execId: " + jobExecuteResult.getExecID() + ", taskId: " + jobExecuteResult.taskID());

79. // 4. 获取脚本的执行状态
80. JobInfoResult jobInfoResult = client.getJobInfo(jobExecuteResult);
81. int sleepTimeMills = 1000;
82. while(!jobInfoResult.isCompleted()) {
83. // 5. 获取脚本的执行进度
84. JobProgressResult progress = client.progress(jobExecuteResult);
85. Utils.sleepQuietly(sleepTimeMills);
86. jobInfoResult = client.getJobInfo(jobExecuteResult);
87. }

90. // 6. 获取脚本的Job信息
91. JobInfoResult jobInfo = client.getJobInfo(jobExecuteResult);
92. // 7. 获取结果集列表（如果用户一次提交多个SQL，会产生多个结果集）
93. String resultSet = jobInfo.getResultSetList(client)[0];
94. // 8. 通过一个结果集信息，获取具体的结果集
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 测试类

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. 配置DWSClientBuilder，通过DWSClientBuilder获取一个DWSClientConfig
29. val clientConfig = DWSClientConfigBuilder.newBuilder()
30. .addServerUrl("http://${ip}:${port}") //指定ServerUrl，Linkis服务器端网关的地址,如http://{ip}:{port}
31. .connectionTimeout(30000)  //connectionTimeOut 客户端连接超时时间
32. .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES)  //是否启用注册发现，如果启用，会自动发现新启动的Gateway
33. .loadbalancerEnabled(true)  // 是否启用负载均衡，如果不启用注册发现，则负载均衡没有意义
34. .maxConnectionSize(5)   //指定最大连接数，即最大并发数
35. .retryEnabled(false).readTimeout(30000)   //执行失败，是否允许重试
36. .setAuthenticationStrategy(new StaticAuthenticationStrategy())  //AuthenticationStrategy Linkis认证方式
37. .setAuthTokenKey("${username}").setAuthTokenValue("${password}") //认证key，一般为用户名;  认证value，一般为用户名对应的密码
38. .setDWSVersion("v1").build()  //Linkis后台协议的版本，当前版本为v1

41. // 2. 通过DWSClientConfig获取一个UJESClient
42. val client = UJESClient(clientConfig)

45. try {
46. // 3. 开始执行代码
47. println("user : " + user + ", code : [" + executeCode + "]")
48. val startupMap = new java.util.HashMap[String, Any]()
49. startupMap.put("wds.linkis.yarnqueue", "q02") //启动参数配置
50. //指定Label
51. val labels: util.Map[String, Any] = new util.HashMap[String, Any]
52. //添加本次执行所依赖的的标签，如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. //指定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提交用户
63. .addExecuteUser(user) //实际执行用户
64. .setLabels(labels)
65. .setSource(source)
66. .build)  //User，请求用户；用于做用户级多租户隔离
67. println("execId: " + jobExecuteResult.getExecID + ", taskId: " + jobExecuteResult.taskID)

70. // 4. 获取脚本的执行状态
71. var jobInfoResult = client.getJobInfo(jobExecuteResult)
72. val sleepTimeMills : Int = 1000
73. while (!jobInfoResult.isCompleted) {
74. // 5. 获取脚本的执行进度
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. 获取脚本的Job信息
88. val jobInfo = client.getJobInfo(jobExecuteResult)
89. // 7. 获取结果集列表（如果用户一次提交多个SQL，会产生多个结果集）
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. 通过一个结果集信息，获取具体的结果集
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. }