0.X SDK的使用
Linkis 提供了方便的JAVA和SCALA调用的接口，只需要引入ujes-client的模块就可以进行使用
1 引入依赖模块#
<dependency> <groupId>com.webank.wedatasphere.linkis</groupId> <artifactId>linkis-ujes-client</artifactId> <version>0.9.3</version></dependency>2 Java测试代码#
建立Java的测试类UJESClientImplTestJ，具体接口含义可以见注释：
package com.wedatasphere.linkis.ujes.test;
import com.webank.wedatasphere.linkis.common.utils.Utils;import com.webank.wedatasphere.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy;import com.webank.wedatasphere.linkis.httpclient.dws.authentication.TokenAuthenticationStrategy;import com.webank.wedatasphere.linkis.httpclient.dws.config.DWSClientConfig;import com.webank.wedatasphere.linkis.httpclient.dws.config.DWSClientConfigBuilder;import com.webank.wedatasphere.linkis.ujes.client.UJESClient;import com.webank.wedatasphere.linkis.ujes.client.UJESClientImpl;import com.webank.wedatasphere.linkis.ujes.client.request.JobExecuteAction;import com.webank.wedatasphere.linkis.ujes.client.request.ResultSetAction;import com.webank.wedatasphere.linkis.ujes.client.response.JobExecuteResult;import com.webank.wedatasphere.linkis.ujes.client.response.JobInfoResult;import com.webank.wedatasphere.linkis.ujes.client.response.JobProgressResult;import org.apache.commons.io.IOUtils;
import java.util.HashMap;import java.util.Map;import java.util.concurrent.TimeUnit;
public class UJESClientTest {
public static void main(String[] args){
String user = "hadoop"; String executeCode = "show databases;";
// 1. 配置DWSClientBuilder，通过DWSClientBuilder获取一个DWSClientConfig DWSClientConfig clientConfig = ((DWSClientConfigBuilder) (DWSClientConfigBuilder.newBuilder() .addUJESServerUrl("http://${ip}:${port}") //指定ServerUrl，linkis服务器端网关的地址,如http://{ip}:{port} .connectionTimeout(30000) //connectionTimeOut 客户端连接超时时间 .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES) //是否启用注册发现，如果启用，会自动发现新启动的Gateway .loadbalancerEnabled(true) // 是否启用负载均衡，如果不启用注册发现，则负载均衡没有意义 .maxConnectionSize(5) //指定最大连接数，即最大并发数 .retryEnabled(false).readTimeout(30000) //执行失败，是否允许重试 .setAuthenticationStrategy(new StaticAuthenticationStrategy()) //AuthenticationStrategy Linkis认证方式 .setAuthTokenKey("${username}").setAuthTokenValue("${password}"))) //认证key，一般为用户名; 认证value，一般为用户名对应的密码 .setDWSVersion("v1").build(); //linkis后台协议的版本，当前版本为v1
// 2. 通过DWSClientConfig获取一个UJESClient UJESClient client = new UJESClientImpl(clientConfig);
try { // 3. 开始执行代码 System.out.println("user : " + user + ", code : [" + executeCode + "]"); Map<String, Object> startupMap = new HashMap<String, Object>(); startupMap.put("wds.linkis.yarnqueue", "default"); // 在startupMap可以存放多种启动参数，参见linkis管理台配置 JobExecuteResult jobExecuteResult = client.execute(JobExecuteAction.builder() .setCreator("linkisClient-Test") //creator，请求linkis的客户端的系统名，用于做系统级隔离 .addExecuteCode(executeCode) //ExecutionCode 请求执行的代码 .setEngineType((JobExecuteAction.EngineType) JobExecuteAction.EngineType$.MODULE$.HIVE()) // 希望请求的linkis的执行引擎类型，如Spark hive等 .setUser(user) //User，请求用户；用于做用户级多租户隔离 .setStartupParams(startupMap) .build()); System.out.println("execId: " + jobExecuteResult.getExecID() + ", taskId: " + jobExecuteResult.taskID());
// 4. 获取脚本的执行状态 JobInfoResult jobInfoResult = client.getJobInfo(jobExecuteResult); int sleepTimeMills = 1000; while(!jobInfoResult.isCompleted()) { // 5. 获取脚本的执行进度 JobProgressResult progress = client.progress(jobExecuteResult); Utils.sleepQuietly(sleepTimeMills); jobInfoResult = client.getJobInfo(jobExecuteResult); }
// 6. 获取脚本的Job信息 JobInfoResult jobInfo = client.getJobInfo(jobExecuteResult); // 7. 获取结果集列表（如果用户一次提交多个SQL，会产生多个结果集） String resultSet = jobInfo.getResultSetList(client)[0]; // 8. 通过一个结果集信息，获取具体的结果集 Object fileContents = client.resultSet(ResultSetAction.builder().setPath(resultSet).setUser(jobExecuteResult.getUser()).build()).getFileContent(); System.out.println("fileContents: " + fileContents);
} catch (Exception e) { e.printStackTrace(); IOUtils.closeQuietly(client); } IOUtils.closeQuietly(client); }}运行上述的代码即可以和Linkis进行交互
3 Scala测试代码#
package com.wedatasphere.linkis.ujes.test
import java.util.concurrent.TimeUnit
import com.webank.wedatasphere.linkis.common.utils.Utilsimport com.webank.wedatasphere.linkis.httpclient.dws.authentication.StaticAuthenticationStrategyimport com.webank.wedatasphere.linkis.httpclient.dws.config.DWSClientConfigBuilderimport com.webank.wedatasphere.linkis.ujes.client.UJESClientimport com.webank.wedatasphere.linkis.ujes.client.request.JobExecuteAction.EngineTypeimport com.webank.wedatasphere.linkis.ujes.client.request.{JobExecuteAction, ResultSetAction}import org.apache.commons.io.IOUtils
object UJESClientImplTest extends App {
var executeCode = "show databases;" var user = "hadoop"
// 1. 配置DWSClientBuilder，通过DWSClientBuilder获取一个DWSClientConfig val clientConfig = DWSClientConfigBuilder.newBuilder() .addUJESServerUrl("http://${ip}:${port}") //指定ServerUrl，Linkis服务器端网关的地址,如http://{ip}:{port} .connectionTimeout(30000) //connectionTimeOut 客户端连接超时时间 .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES) //是否启用注册发现，如果启用，会自动发现新启动的Gateway .loadbalancerEnabled(true) // 是否启用负载均衡，如果不启用注册发现，则负载均衡没有意义 .maxConnectionSize(5) //指定最大连接数，即最大并发数 .retryEnabled(false).readTimeout(30000) //执行失败，是否允许重试 .setAuthenticationStrategy(new StaticAuthenticationStrategy()) //AuthenticationStrategy Linkis认证方式 .setAuthTokenKey("${username}").setAuthTokenValue("${password}") //认证key，一般为用户名; 认证value，一般为用户名对应的密码 .setDWSVersion("v1").build() //Linkis后台协议的版本，当前版本为v1
// 2. 通过DWSClientConfig获取一个UJESClient val client = UJESClient(clientConfig) try { // 3. 开始执行代码 println("user : " + user + ", code : [" + executeCode + "]") val startupMap = new java.util.HashMap[String, Any]() startupMap.put("wds.linkis.yarnqueue", "default") //启动参数配置 val jobExecuteResult = client.execute(JobExecuteAction.builder() .setCreator("LinkisClient-Test") //creator，请求Linkis的客户端的系统名，用于做系统级隔离 .addExecuteCode(executeCode) //ExecutionCode 请求执行的代码 .setEngineType(EngineType.SPARK) // 希望请求的Linkis的执行引擎类型，如Spark hive等 .setStartupParams(startupMap) .setUser(user).build()) //User，请求用户；用于做用户级多租户隔离 println("execId: " + jobExecuteResult.getExecID + ", taskId: " + jobExecuteResult.taskID) // 4. 获取脚本的执行状态 var jobInfoResult = client.getJobInfo(jobExecuteResult) val sleepTimeMills : Int = 1000 while (!jobInfoResult.isCompleted) { // 5. 获取脚本的执行进度 val progress = client.progress(jobExecuteResult) val progressInfo = if (progress.getProgressInfo != null) progress.getProgressInfo.toList else List.empty println("progress: " + progress.getProgress + ", progressInfo: " + progressInfo) Utils.sleepQuietly(sleepTimeMills) jobInfoResult = client.getJobInfo(jobExecuteResult) } if (!jobInfoResult.isSucceed) { println("Failed to execute job: " + jobInfoResult.getMessage) throw new Exception(jobInfoResult.getMessage) }
// 6. 获取脚本的Job信息 val jobInfo = client.getJobInfo(jobExecuteResult) // 7. 获取结果集列表（如果用户一次提交多个SQL，会产生多个结果集） val resultSetList = jobInfoResult.getResultSetList(client) println("All result set list:") resultSetList.foreach(println) val oneResultSet = jobInfo.getResultSetList(client).head // 8. 通过一个结果集信息，获取具体的结果集 val fileContents = client.resultSet(ResultSetAction.builder().setPath(oneResultSet).setUser(jobExecuteResult.getUser).build()).getFileContent println("First fileContents: ") println(fileContents) } catch { case e: Exception => { e.printStackTrace() } } IOUtils.closeQuietly(client)}