• Linkis快速使用文档
    • 1.概述
    • 2 快速运行
    • 3 快速实现
      • 3.1 maven依赖
    • 3.2 参考实现

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