Linkis快速使用文档

1.概述

Linkis为用户提供了Java客户端的实现，用户可以使用UJESClient对Linkis后台服务实现快速访问。

2 快速运行

我们在ujes/client/src/test模块下，提供了UJESClient的两个测试类：

  1.     com.webank.wedatasphere.linkis.ujes.client.UJESClientImplTestJ # 基于Java实现的测试类
  2.     com.webank.wedatasphere.linkis.ujes.client.UJESClientImplTest # 基于Scala实现的测试类

如果您clone了Linkis的源代码，可以直接运行这两个测试类。

下面具体介绍如何快速实现一次对Linkis的代码提交执行。

3 快速实现

3.1 maven依赖
  1. <dependency>
  2.   <groupId>com.webank.wedatasphere.Linkis</groupId>
  3.   <artifactId>Linkis-ujes-client</artifactId>
  4.   <version>0.6.0</version>
  5. </dependency>

3.2 参考实现

  • JAVA
  1. package com.webank.bdp.dataworkcloud.ujes.client;
  3. import com.webank.wedatasphere.Linkis.common.utils.Utils;
  4. import com.webank.wedatasphere.Linkis.httpclient.dws.authentication.StaticAuthenticationStrategy;
  5. import com.webank.wedatasphere.Linkis.httpclient.dws.config.DWSClientConfig;
  6. import com.webank.wedatasphere.Linkis.httpclient.dws.config.DWSClientConfigBuilder;
  7. import com.webank.wedatasphere.Linkis.ujes.client.UJESClient;
  8. import com.webank.wedatasphere.Linkis.ujes.client.UJESClientImpl;
  9. import com.webank.wedatasphere.Linkis.ujes.client.request.JobExecuteAction;
  10. import com.webank.wedatasphere.Linkis.ujes.client.request.ResultSetAction;
  11. import com.webank.wedatasphere.Linkis.ujes.client.response.JobExecuteResult;
  12. import com.webank.wedatasphere.Linkis.ujes.client.response.JobInfoResult;
  13. import com.webank.wedatasphere.Linkis.ujes.client.response.JobProgressResult;
  14. import com.webank.wedatasphere.Linkis.ujes.client.response.JobStatusResult;
  15. import org.apache.commons.io.IOUtils;
  17. import java.util.concurrent.TimeUnit;
  20. public class UJESClientImplTestJ{
  21.     public static void main(String[] args){
  22.         // 1. 配置DWSClientBuilder，通过DWSClientBuilder获取一个DWSClientConfig
  23.         DWSClientConfig clientConfig = ((DWSClientConfigBuilder) (DWSClientConfigBuilder.newBuilder()
  24.                 .addUJESServerUrl("http://${ip}:${port}")  //指定ServerUrl，Linkis服务器端网关的地址,如http://{ip}:{port}
  25.                 .connectionTimeout(30000)   //connectionTimeOut 客户端连接超时时间
  26.                 .discoveryEnabled(true).discoveryFrequency(1, TimeUnit.MINUTES)  //是否启用注册发现，如果启用，会自动发现新启动的Gateway
  27.                 .loadbalancerEnabled(true)  // 是否启用负载均衡，如果不启用注册发现，则负载均衡没有意义
  28.                 .maxConnectionSize(5)   //指定最大连接数，即最大并发数
  29.                 .retryEnabled(false).readTimeout(30000)   //执行失败，是否允许重试
  30.                 .setAuthenticationStrategy(new StaticAuthenticationStrategy())   //AuthenticationStrategy Linkis认证方式
  31.                 .setAuthTokenKey("johnnwang").setAuthTokenValue("Abcd1234")))  //认证key，一般为用户名;  认证value，一般为用户名对应的密码
  32.                 .setDWSVersion("v1").build();  //Linkis后台协议的版本，当前版本为v1
  34.         // 2. 通过DWSClientConfig获取一个UJESClient
  35.         UJESClient client = new UJESClientImpl(clientConfig);
  37.         // 3. 开始执行代码
  38.         JobExecuteResult jobExecuteResult = client.execute(JobExecuteAction.builder()
  39.                 .setCreator("LinkisClient-Test")  //creator，请求Linkis的客户端的系统名，用于做系统级隔离
  40.                 .addExecuteCode("show tables")   //ExecutionCode 请求执行的代码
  41.                 .setEngineType(JobExecuteAction.EngineType$.MODULE$.HIVE()) // 希望请求的Linkis的执行引擎类型，如Spark hive等
  42.                 .setUser("johnnwang")   //User，请求用户；用于做用户级多租户隔离
  43.                 .build());
  44.         System.out.println("execId: " + jobExecuteResult.getExecID() + ", taskId: " + jobExecuteResult.taskID());
  46.         // 4. 获取脚本的执行状态
  47.         JobStatusResult status = client.status(jobExecuteResult);
  48.         while(!status.isCompleted()) {
  49.             // 5. 获取脚本的执行进度
  50.             JobProgressResult progress = client.progress(jobExecuteResult);
  51.             Utils.sleepQuietly(500);
  52.             status = client.status(jobExecuteResult);
  53.         }
  55.         // 6. 获取脚本的Job信息
  56.         JobInfoResult jobInfo = client.getJobInfo(jobExecuteResult);
  57.         // 7. 获取结果集列表（如果用户一次提交多个SQL，会产生多个结果集）
  58.         String resultSet = jobInfo.getResultSetList(client)[0];
  59.         // 8. 通过一个结果集信息，获取具体的结果集
  60.         Object fileContents = client.resultSet(ResultSetAction.builder().setPath(resultSet).setUser(jobExecuteResult.getUser()).build()).getFileContent();
  61.         System.out.println("fileContents: " + fileContents);
  62.         IOUtils.closeQuietly(client);
  63.     }
  64. }
  • SCALA
  2. import java.util.concurrent.TimeUnit
  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.DWSClientConfigBuilder
  7. import com.webank.wedatasphere.Linkis.ujes.client.request.JobExecuteAction.EngineType
  8. import com.webank.wedatasphere.Linkis.ujes.client.request.{JobExecuteAction, ResultSetAction}
  9. import org.apache.commons.io.IOUtils
  11. object UJESClientImplTest extends App {
  13.   // 1. 配置DWSClientBuilder，通过DWSClientBuilder获取一个DWSClientConfig
  14.   val clientConfig = DWSClientConfigBuilder.newBuilder()
  15.     .addUJESServerUrl("http://${ip}:${port}")  //指定ServerUrl，Linkis服务器端网关的地址,如http://{ip}:{port}
  16.     .connectionTimeout(30000)  //connectionTimeOut 客户端连接超时时间
  17.     .discoveryEnabled(true).discoveryFrequency(1, TimeUnit.MINUTES)  //是否启用注册发现，如果启用，会自动发现新启动的Gateway
  18.     .loadbalancerEnabled(true)  // 是否启用负载均衡，如果不启用注册发现，则负载均衡没有意义
  19.     .maxConnectionSize(5)   //指定最大连接数，即最大并发数
  20.     .retryEnabled(false).readTimeout(30000)   //执行失败，是否允许重试
  21.     .setAuthenticationStrategy(new StaticAuthenticationStrategy())  //AuthenticationStrategy Linkis认证方式
  22.     .setAuthTokenKey("${username}").setAuthTokenValue("${password}")  //认证key，一般为用户名;  认证value，一般为用户名对应的密码
  23.     .setDWSVersion("v1").build()  //Linkis后台协议的版本，当前版本为v1
  25.   // 2. 通过DWSClientConfig获取一个UJESClient
  26.   val client = UJESClient(clientConfig)
  28.   // 3. 开始执行代码
  29.   val jobExecuteResult = client.execute(JobExecuteAction.builder()
  30.     .setCreator("LinkisClient-Test")  //creator，请求Linkis的客户端的系统名，用于做系统级隔离
  31.     .addExecuteCode("show tables")   //ExecutionCode 请求执行的代码
  32.     .setEngineType(EngineType.SPARK) // 希望请求的Linkis的执行引擎类型，如Spark hive等
  33.     .setUser("${username}").build())  //User，请求用户；用于做用户级多租户隔离
  34.   println("execId: " + jobExecuteResult.getExecID + ", taskId: " + jobExecuteResult.taskID)
  36.   // 4. 获取脚本的执行状态
  37.   var status = client.status(jobExecuteResult)
  38.   while(!status.isCompleted) {
  39.   // 5. 获取脚本的执行进度
  40.     val progress = client.progress(jobExecuteResult)
  41.     val progressInfo = if(progress.getProgressInfo != null) progress.getProgressInfo.toList else List.empty
  42.     println("progress: " + progress.getProgress + ", progressInfo: " + progressInfo)
  43.     Utils.sleepQuietly(500)
  44.     status = client.status(jobExecuteResult)
  45.   }
  47.   // 6. 获取脚本的Job信息
  48.   val jobInfo = client.getJobInfo(jobExecuteResult)
  49.   // 7. 获取结果集列表（如果用户一次提交多个SQL，会产生多个结果集）
  50.   val resultSet = jobInfo.getResultSetList(client).head
  51.   // 8. 通过一个结果集信息，获取具体的结果集
  52.   val fileContents = client.resultSet(ResultSetAction.builder().setPath(resultSet).setUser(jobExecuteResult.getUser).build()).getFileContent
  53.   println("fileContents: " + fileContents)
  54.   IOUtils.closeQuietly(client)
  55. }