JAVA SDK 方式使用

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

1. 引入依赖模块

  1. <dependency>
  2.   <groupId>org.apache.linkis</groupId>
  3.   <artifactId>linkis-computation-client</artifactId>
  4.   <version>${linkis.version}</version>
  5. </dependency>
  6. 如：
  7. <dependency>
  8.   <groupId>org.apache.linkis</groupId>
  9.   <artifactId>linkis-computation-client</artifactId>
  10.   <version>1.0.3</version>
  11. </dependency>

2. Java测试代码

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

  1. package org.apache.linkis.client.test;
  4. import org.apache.linkis.common.utils.Utils;
  5. import org.apache.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy;
  6. import org.apache.linkis.httpclient.dws.config.DWSClientConfig;
  7. import org.apache.linkis.httpclient.dws.config.DWSClientConfigBuilder;
  8. import org.apache.linkis.manager.label.constant.LabelKeyConstant;
  9. import org.apache.linkis.protocol.constants.TaskConstant;
  10. import org.apache.linkis.ujes.client.UJESClient;
  11. import org.apache.linkis.ujes.client.UJESClientImpl;
  12. import org.apache.linkis.ujes.client.request.JobSubmitAction;
  13. import org.apache.linkis.ujes.client.request.JobExecuteAction;
  14. import org.apache.linkis.ujes.client.request.ResultSetAction;
  15. import org.apache.linkis.ujes.client.response.*;
  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.     // 1. build config: linkis gateway url
  28.     private static DWSClientConfig clientConfig = ((DWSClientConfigBuilder) (DWSClientConfigBuilder.newBuilder()
  29.             .addServerUrl("http://127.0.0.1:9001/")   //set linkis-mg-gateway url: http://{ip}:{port}
  30.             .connectionTimeout(30000)   //connectionTimeOut
  31.             .discoveryEnabled(false) //disable discovery
  32.             .discoveryFrequency(1, TimeUnit.MINUTES)  // discovery frequency
  33.             .loadbalancerEnabled(true)  // enable loadbalance
  34.             .maxConnectionSize(5)   // set max Connection
  35.             .retryEnabled(false) // set retry
  36.             .readTimeout(30000)  //set read timeout
  37.             .setAuthenticationStrategy(new StaticAuthenticationStrategy())   //AuthenticationStrategy Linkis authen suppory static and Token
  38.             .setAuthTokenKey("hadoop")  // set submit user
  39.             .setAuthTokenValue("hadoop")))  // set passwd or token (setAuthTokenValue("test"))
  40.             .setDWSVersion("v1") //linkis rest version v1
  41.             .build();
  44.     // 2. new Client(Linkis Client) by clientConfig
  45.     private static UJESClient client = new UJESClientImpl(clientConfig);
  48.     public static void main(String[] args){
  51.         String user = "hadoop"; // execute user
  52.         String executeCode = "df=spark.sql(\"show tables\")\n" +
  53.                 "show(df)"; // code support:sql/hql/py/scala
  54.         try {
  57.             System.out.println("user : " + user + ", code : [" + executeCode + "]");
  58.             // 3.推荐用submit的方式，可以指定任务相关的label支持更多特性
  59.             JobExecuteResult jobExecuteResult = toSubmit(user, executeCode);
  60.             //0.x兼容的方式，不推荐使用:JobExecuteResult jobExecuteResult = toExecute(user, executeCode);
  61.             System.out.println("execId: " + jobExecuteResult.getExecID() + ", taskId: " + jobExecuteResult.taskID());
  62.             // 4. get job jonfo
  63.             JobInfoResult jobInfoResult = client.getJobInfo(jobExecuteResult);
  64.             int sleepTimeMills = 1000;
  65.             int logFromLen = 0;
  66.             int logSize = 100;
  67.             while(!jobInfoResult.isCompleted()) {
  68.                 // 5. get progress and log
  69.                 JobProgressResult progress = client.progress(jobExecuteResult);
  70.                 System.out.println("progress: " + progress.getProgress());
  71.                 JobLogResult logRes = client.log(jobExecuteResult, logFromLen, logSize);
  72.                 logFromLen = logRes.fromLine();
  73.                 // 0: info 1: warn 2: error 3: all
  74.                 System.out.println(logRes.log().get(3));
  75.                 Utils.sleepQuietly(sleepTimeMills);
  76.                 jobInfoResult = client.getJobInfo(jobExecuteResult);
  77.             }
  80.             JobInfoResult jobInfo = client.getJobInfo(jobExecuteResult);
  81.             // 6. Get the result set list (if the user submits multiple SQLs at a time,
  82.             // multiple result sets will be generated)
  83.             String resultSet = jobInfo.getResultSetList(client)[0];
  84.             // 7. get resultContent
  85.             Object fileContents = client.resultSet(ResultSetAction.builder().setPath(resultSet).setUser(jobExecuteResult.getUser()).build()).getFileContent();
  86.             System.out.println("res: " + fileContents);
  87.         } catch (Exception e) {
  88.             e.printStackTrace();
  89.             IOUtils.closeQuietly(client);
  90.         }
  91.         IOUtils.closeQuietly(client);
  92.     }
  95.     /**
  96.      * Linkis 1.0 recommends the use of Submit method
  97.      */
  98.     private static JobExecuteResult toSubmit(String user, String code) {
  99.         // 1. build  params
  100.         // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
  101.         Map<String, Object> labels = new HashMap<String, Object>();
  102.         labels.put(LabelKeyConstant.ENGINE_TYPE_KEY, "spark-2.4.3"); // required engineType Label
  103.         labels.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, user + "-APPName");// 请求的用户和应用名，两个参数都不能少，其中APPName不能带有"-"建议替换为"_"
  104.         labels.put(LabelKeyConstant.CODE_TYPE_KEY, "py"); // 指定脚本类型
  105.         // set start up map :engineConn start params
  106.         Map<String, Object> startupMap = new HashMap<String, Object>(16);
  107.         // Support setting engine native parameters,For example: parameters of engines such as spark/hive
  108.         startupMap.put("spark.executor.instances", 2);
  109.         // setting linkis params
  110.         startupMap.put("wds.linkis.rm.yarnqueue", "dws");
  113.         // 2. build jobSubmitAction
  114.         JobSubmitAction jobSubmitAction = JobSubmitAction.builder()
  115.                 .addExecuteCode(code)
  116.                 .setStartupParams(startupMap)
  117.                 .setUser(user) //submit user
  118.                 .addExecuteUser(user)  // execute user
  119.                 .setLabels(labels)
  120.                 .build();
  121.         // 3. to execute
  122.         return client.submit(jobSubmitAction);
  123.     }
  126.     /**
  127.      * Compatible with 0.X execution mode
  128.      */
  129.     private static JobExecuteResult toExecute(String user, String code) {
  130.         // 1. build  params
  131.         // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
  132.         Map<String, Object> labels = new HashMap<String, Object>();
  133.         // labels.put(LabelKeyConstant.TENANT_KEY, "fate");
  134.         // set start up map :engineConn start params
  135.         Map<String, Object> startupMap = new HashMap<String, Object>(16);
  136.         // Support setting engine native parameters,For example: parameters of engines such as spark/hive
  137.         startupMap.put("spark.executor.instances", 2);
  138.         // setting linkis params
  139.         startupMap.put("wds.linkis.rm.yarnqueue", "dws");
  142.         // 2. build JobExecuteAction (0.X old way of using)
  143.         JobExecuteAction executionAction = JobExecuteAction.builder()
  144.                 .setCreator("AppName")  //creator, the system name of the client requesting linkis, used for system-level isolation
  145.                 .addExecuteCode(code)   //Execution Code
  146.                 .setEngineTypeStr("spark") // engineConn type
  147.                 .setRunTypeStr("py") // code type
  148.                 .setUser(user)   //execute user
  149.                 .setStartupParams(startupMap) // start up params
  150.                 .build();
  151.         executionAction.addRequestPayload(TaskConstant.LABELS, labels);
  152.         String body = executionAction.getRequestPayload();
  153.         System.out.println(body);
  156.         // 3. to execute
  157.         return client.execute(executionAction);
  158.     }
  159. }

运行上述的代码即可以完成任务提交/执行/日志/结果集获取等

3. Scala测试代码：

  1. package org.apache.linkis.client.test
  4. import java.util
  5. import java.util.concurrent.TimeUnit
  8. import org.apache.linkis.common.utils.Utils
  9. import org.apache.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy
  10. import org.apache.linkis.httpclient.dws.config.DWSClientConfigBuilder
  11. import org.apache.linkis.manager.label.constant.LabelKeyConstant
  12. import org.apache.linkis.protocol.constants.TaskConstant
  13. import org.apache.linkis.ujes.client.UJESClient
  14. import org.apache.linkis.ujes.client.request._
  15. import org.apache.linkis.ujes.client.response._
  16. import org.apache.commons.io.IOUtils
  17. import org.apache.commons.lang.StringUtils
  20. object LinkisClientTest {
  21. // 1. build config: linkis gateway url
  22.   val clientConfig = DWSClientConfigBuilder.newBuilder()
  23.     .addServerUrl("http://127.0.0.1:9001/")   //set linkis-mg-gateway url: http://{ip}:{port}
  24.     .connectionTimeout(30000)   //connectionTimeOut
  25.     .discoveryEnabled(false) //disable discovery
  26.     .discoveryFrequency(1, TimeUnit.MINUTES)  // discovery frequency
  27.     .loadbalancerEnabled(true)  // enable loadbalance
  28.     .maxConnectionSize(5)   // set max Connection
  29.     .retryEnabled(false) // set retry
  30.     .readTimeout(30000)  //set read timeout
  31.     .setAuthenticationStrategy(new StaticAuthenticationStrategy())   //AuthenticationStrategy Linkis authen suppory static and Token
  32.     .setAuthTokenKey("hadoop")  // set submit user
  33.     .setAuthTokenValue("hadoop")  // set passwd or token (setAuthTokenValue("BML-AUTH"))
  34.     .setDWSVersion("v1") //linkis rest version v1
  35.     .build();
  38.   // 2. new Client(Linkis Client) by clientConfig
  39.   val client = UJESClient(clientConfig)
  42.   def main(args: Array[String]): Unit = {
  43.     val user = "hadoop" // execute user
  44.     val executeCode = "df=spark.sql(\"show tables\")\n" +
  45.       "show(df)"; // code support:sql/hql/py/scala
  46.     try {
  47.       // 3. build job and execute
  48.       println("user : " + user + ", code : [" + executeCode + "]")
  49.       //推荐使用submit，支持传递任务label
  50.       val jobExecuteResult = toSubmit(user, executeCode)
  51.       //0.X: val jobExecuteResult = toExecute(user, executeCode)
  52.       println("execId: " + jobExecuteResult.getExecID + ", taskId: " + jobExecuteResult.taskID)
  53.       // 4. get job jonfo
  54.       var jobInfoResult = client.getJobInfo(jobExecuteResult)
  55.       var logFromLen = 0
  56.       val logSize = 100
  57.       val sleepTimeMills : Int = 1000
  58.       while (!jobInfoResult.isCompleted) {
  59.         // 5. get progress and log
  60.         val progress = client.progress(jobExecuteResult)
  61.        println("progress: " + progress.getProgress)
  62.         val logObj = client .log(jobExecuteResult, logFromLen, logSize)
  63.         logFromLen = logObj.fromLine
  64.         val logArray = logObj.getLog
  65.         // 0: info 1: warn 2: error 3: all
  66.         if (logArray != null && logArray.size >= 4 && StringUtils.isNotEmpty(logArray.get(3))) {
  67.           println(s"log: ${logArray.get(3)}")
  68.         }
  69.         Utils.sleepQuietly(sleepTimeMills)
  70.         jobInfoResult = client.getJobInfo(jobExecuteResult)
  71.       }
  72.       if (!jobInfoResult.isSucceed) {
  73.         println("Failed to execute job: " + jobInfoResult.getMessage)
  74.         throw new Exception(jobInfoResult.getMessage)
  75.       }
  78.       // 6. Get the result set list (if the user submits multiple SQLs at a time,
  79.       // multiple result sets will be generated)
  80.       val jobInfo = client.getJobInfo(jobExecuteResult)
  81.       val resultSetList = jobInfoResult.getResultSetList(client)
  82.       println("All result set list:")
  83.       resultSetList.foreach(println)
  84.       val oneResultSet = jobInfo.getResultSetList(client).head
  85.       // 7. get resultContent
  86.       val fileContents = client.resultSet(ResultSetAction.builder().setPath(oneResultSet).setUser(jobExecuteResult.getUser).build()).getFileContent
  87.       println("First fileContents: ")
  88.       println(fileContents)
  89.     } catch {
  90.       case e: Exception => {
  91.         e.printStackTrace()
  92.       }
  93.     }
  94.     IOUtils.closeQuietly(client)
  95.   }
  98.   /**
  99.    * Linkis 1.0 recommends the use of Submit method
  100.    */
  101.   def toSubmit(user: String, code: String): JobExecuteResult = {
  102.     // 1. build  params
  103.     // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
  104.     val labels: util.Map[String, Any] = new util.HashMap[String, Any]
  105.     labels.put(LabelKeyConstant.ENGINE_TYPE_KEY, "spark-2.4.3"); // required engineType Label
  106.     labels.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, user + "-APPName");// 请求的用户和应用名，两个参数都不能少，其中APPName不能带有"-"建议替换为"_"
  107.     labels.put(LabelKeyConstant.CODE_TYPE_KEY, "py"); // 指定脚本类型
  110.     val startupMap = new java.util.HashMap[String, Any]()
  111.     // Support setting engine native parameters,For example: parameters of engines such as spark/hive
  112.     startupMap.put("spark.executor.instances", 2);
  113.     // setting linkis params
  114.     startupMap.put("wds.linkis.rm.yarnqueue", "default");
  115.     // 2. build jobSubmitAction
  116.     val jobSubmitAction = JobSubmitAction.builder
  117.       .addExecuteCode(code)
  118.       .setStartupParams(startupMap)
  119.       .setUser(user) //submit user
  120.       .addExecuteUser(user) //execute user
  121.       .setLabels(labels)
  122.       .build
  123.     // 3. to execute
  124.     client.submit(jobSubmitAction)
  125.   }
  130.   /**
  131.    * Compatible with 0.X execution mode
  132.    */
  133.   def toExecute(user: String, code: String): JobExecuteResult = {
  134.     // 1. build  params
  135.     // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
  136.     val labels = new util.HashMap[String, Any]
  137.     // labels.put(LabelKeyConstant.TENANT_KEY, "fate");
  140.     val startupMap = new java.util.HashMap[String, Any]()
  141.     // Support setting engine native parameters,For example: parameters of engines such as spark/hive
  142.     startupMap.put("spark.executor.instances", 2)
  143.     // setting linkis params
  144.     startupMap.put("wds.linkis.rm.yarnqueue", "default")
  145.     // 2. build JobExecuteAction (0.X old way of using)
  146.     val  executionAction = JobExecuteAction.builder()
  147.       .setCreator("APPName")  //creator, the system name of the client requesting linkis, used for system-level isolation
  148.       .addExecuteCode(code)   //Execution Code
  149.       .setEngineTypeStr("spark") // engineConn type
  150.       .setRunTypeStr("py") // code type
  151.       .setUser(user)   //execute user
  152.       .setStartupParams(startupMap) // start up params
  153.       .build();
  154.     executionAction.addRequestPayload(TaskConstant.LABELS, labels);
  155.     // 3. to execute
  156.     client.execute(executionAction)
  157.   }
  162. }