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