JAVA SDK 的使用

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

  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>

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

  1. package org.apache.linkis.client.test;
  3. import org.apache.linkis.common.utils.Utils;
  4. import org.apache.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy;
  5. import org.apache.linkis.httpclient.dws.config.DWSClientConfig;
  6. import org.apache.linkis.httpclient.dws.config.DWSClientConfigBuilder;
  7. import org.apache.linkis.manager.label.constant.LabelKeyConstant;
  8. import org.apache.linkis.protocol.constants.TaskConstant;
  9. import org.apache.linkis.ujes.client.UJESClient;
  10. import org.apache.linkis.ujes.client.UJESClientImpl;
  11. import org.apache.linkis.ujes.client.request.JobSubmitAction;
  12. import org.apache.linkis.ujes.client.request.JobExecuteAction;
  13. import org.apache.linkis.ujes.client.request.ResultSetAction;
  14. import org.apache.linkis.ujes.client.response.*;
  15. import org.apache.commons.io.IOUtils;
  17. import java.util.HashMap;
  18. import java.util.Map;
  19. import java.util.concurrent.TimeUnit;
  21. public class LinkisClientTest {
  23.     // 1. build config: linkis gateway url
  24.     private static DWSClientConfig clientConfig = ((DWSClientConfigBuilder) (DWSClientConfigBuilder.newBuilder()
  25.             .addServerUrl("http://127.0.0.1:9001/")   //set linkis-mg-gateway url: http://{ip}:{port}
  26.             .connectionTimeout(30000)   //connectionTimeOut
  27.             .discoveryEnabled(false) //disable discovery
  28.             .discoveryFrequency(1, TimeUnit.MINUTES)  // discovery frequency
  29.             .loadbalancerEnabled(true)  // enable loadbalance
  30.             .maxConnectionSize(5)   // set max Connection
  31.             .retryEnabled(true) // set retry
  32.             .readTimeout(30000)  //set read timeout
  33.             .setAuthenticationStrategy(new StaticAuthenticationStrategy())   //AuthenticationStrategy Linkis authen suppory static and Token
  34.             .setAuthTokenKey("hadoop")  // set submit user
  35.             .setAuthTokenValue("123456")))  // set passwd or token (setAuthTokenValue("test"))
  36.             .setDWSVersion("v1") //linkis rest version v1
  37.             .build();
  39.     // 2. new Client(Linkis Client) by clientConfig
  40.     private static UJESClient client = new UJESClientImpl(clientConfig);
  42.     public static void main(String[] args) {
  44.         String user = "hadoop"; // 用户需要和AuthTokenKey的值保持一致
  45.         String executeCode = "df=spark.sql(\"show tables\")\n" +
  46.                 "show(df)"; // code support:sql/hql/py/scala
  47.         try {
  49.             System.out.println("user : " + user + ", code : [" + executeCode + "]");
  50.             // 3. build job and execute
  51.             JobExecuteResult jobExecuteResult = toSubmit(user, executeCode);
  52.             System.out.println("execId: " + jobExecuteResult.getExecID() + ", taskId: " + jobExecuteResult.taskID());
  53.             // 4. get job jonfo
  54.             JobInfoResult jobInfoResult = client.getJobInfo(jobExecuteResult);
  55.             int sleepTimeMills = 1000;
  56.             int logFromLen = 0;
  57.             int logSize = 100;
  58.             while (!jobInfoResult.isCompleted()) {
  59.                 // 5. get progress and log
  60.                 JobProgressResult progress = client.progress(jobExecuteResult);
  61.                 System.out.println("progress: " + progress.getProgress());
  62.                 JobLogResult logRes = client.log(jobExecuteResult, logFromLen, logSize);
  63.                 logFromLen = logRes.fromLine();
  64.                 // 0: info 1: warn 2: error 3: all
  65.                 System.out.println(logRes.log().get(3));
  66.                 Utils.sleepQuietly(sleepTimeMills);
  67.                 jobInfoResult = client.getJobInfo(jobExecuteResult);
  68.             }
  70.             JobInfoResult jobInfo = client.getJobInfo(jobExecuteResult);
  71.             // 6. Get the result set list (if the user submits multiple SQLs at a time,
  72.             // multiple result sets will be generated)
  73.             String resultSet = jobInfo.getResultSetList(client)[0];
  74.             // 7. get resultContent
  75.             ResultSetResult resultSetResult = client.resultSet(ResultSetAction.builder().setPath(resultSet).setUser(jobExecuteResult.getUser()).build());
  76.             System.out.println("metadata: " + resultSetResult.getMetadata()); // column name type
  77.             System.out.println("res: " + resultSetResult.getFileContent()); //row data
  78.         } catch (Exception e) {
  79.             e.printStackTrace();// please use log
  80.             IOUtils.closeQuietly(client);
  81.         }
  82.         IOUtils.closeQuietly(client);
  83.     }
  86.     private static JobExecuteResult toSubmit(String user, String code) {
  87.         // 1. build  params
  88.         // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
  89.         Map<String, Object> labels = new HashMap<String, Object>();
  90.         labels.put(LabelKeyConstant.ENGINE_TYPE_KEY, "spark-2.4.3"); // required engineType Label
  91.         labels.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, user + "-APPName");// required execute user and creator eg:hadoop-IDE
  92.         labels.put(LabelKeyConstant.CODE_TYPE_KEY, "py"); // required codeType
  93.         // set start up map :engineConn start params
  94.         Map<String, Object> startupMap = new HashMap<String, Object>(16);
  95.         // Support setting engine native parameters,For example: parameters of engines such as spark/hive
  96.         startupMap.put("spark.executor.instances", 2);
  97.         // setting linkis params
  98.         startupMap.put("wds.linkis.rm.yarnqueue", "dws");
  100.         // 2. build jobSubmitAction
  101.         JobSubmitAction jobSubmitAction = JobSubmitAction.builder()
  102.                 .addExecuteCode(code)
  103.                 .setStartupParams(startupMap)
  104.                 .setUser(user) //submit user
  105.                 .addExecuteUser(user)  // execute user
  106.                 .setLabels(labels)
  107.                 .build();
  108.         // 3. to execute
  109.         return client.submit(jobSubmitAction);
  110.     }
  111. }

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

  1. package org.apache.linkis.client.test
  3. import org.apache.commons.io.IOUtils
  4. import org.apache.commons.lang3.StringUtils
  5. import org.apache.linkis.common.utils.Utils
  6. import org.apache.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy
  7. import org.apache.linkis.httpclient.dws.config.DWSClientConfigBuilder
  8. import org.apache.linkis.manager.label.constant.LabelKeyConstant
  9. import org.apache.linkis.ujes.client.request._
  10. import org.apache.linkis.ujes.client.response._
  12. import java.util
  13. import java.util.concurrent.TimeUnit
  15. object LinkisClientTest {
  16.   // 1. build config: linkis gateway url
  17.   val clientConfig = DWSClientConfigBuilder.newBuilder()
  18.     .addServerUrl("http://127.0.0.1:9001/") //set linkis-mg-gateway url: http://{ip}:{port}
  19.     .connectionTimeout(30000) //connectionTimeOut
  20.     .discoveryEnabled(false) //disable discovery
  21.     .discoveryFrequency(1, TimeUnit.MINUTES) // discovery frequency
  22.     .loadbalancerEnabled(true) // enable loadbalance
  23.     .maxConnectionSize(5) // set max Connection
  24.     .retryEnabled(false) // set retry
  25.     .readTimeout(30000) //set read timeout
  26.     .setAuthenticationStrategy(new StaticAuthenticationStrategy()) //AuthenticationStrategy Linkis authen suppory static and Token
  27.     .setAuthTokenKey("hadoop") // set submit user
  28.     .setAuthTokenValue("hadoop") // set passwd or token (setAuthTokenValue("BML-AUTH"))
  29.     .setDWSVersion("v1") //linkis rest version v1
  30.     .build();
  32.   // 2. new Client(Linkis Client) by clientConfig
  33.   val client = UJESClient(clientConfig)
  35.   def main(args: Array[String]): Unit = {
  36.     val user = "hadoop" // execute user 用户需要和AuthTokenKey的值保持一致
  37.     val executeCode = "df=spark.sql(\"show tables\")\n" +
  38.       "show(df)"; // code support:sql/hql/py/scala
  39.     try {
  40.       // 3. build job and execute
  41.       println("user : " + user + ", code : [" + executeCode + "]")
  42.       // 推荐使用submit，支持传递任务label
  43.       val jobExecuteResult = toSubmit(user, executeCode)
  44.       println("execId: " + jobExecuteResult.getExecID + ", taskId: " + jobExecuteResult.taskID)
  45.       // 4. get job jonfo
  46.       var jobInfoResult = client.getJobInfo(jobExecuteResult)
  47.       var logFromLen = 0
  48.       val logSize = 100
  49.       val sleepTimeMills: Int = 1000
  50.       while (!jobInfoResult.isCompleted) {
  51.         // 5. get progress and log
  52.         val progress = client.progress(jobExecuteResult)
  53.         println("progress: " + progress.getProgress)
  54.         val logObj = client.log(jobExecuteResult, logFromLen, logSize)
  55.         logFromLen = logObj.fromLine
  56.         val logArray = logObj.getLog
  57.         // 0: info 1: warn 2: error 3: all
  58.         if (logArray != null && logArray.size >= 4 && StringUtils.isNotEmpty(logArray.get(3))) {
  59.           println(s"log: ${logArray.get(3)}")
  60.         }
  61.         Utils.sleepQuietly(sleepTimeMills)
  62.         jobInfoResult = client.getJobInfo(jobExecuteResult)
  63.       }
  64.       if (!jobInfoResult.isSucceed) {
  65.         println("Failed to execute job: " + jobInfoResult.getMessage)
  66.         throw new Exception(jobInfoResult.getMessage)
  67.       }
  69.       // 6. Get the result set list (if the user submits multiple SQLs at a time,
  70.       // multiple result sets will be generated)
  71.       val jobInfo = client.getJobInfo(jobExecuteResult)
  72.       val resultSetList = jobInfoResult.getResultSetList(client)
  73.       println("All result set list:")
  74.       resultSetList.foreach(println)
  75.       val oneResultSet = jobInfo.getResultSetList(client).head
  76.       // 7. get resultContent
  77.       val resultSetResult: ResultSetResult = client.resultSet(ResultSetAction.builder.setPath(oneResultSet).setUser(jobExecuteResult.getUser).build)
  78.       println("metadata: " + resultSetResult.getMetadata) // column name type
  79.       println("res: " + resultSetResult.getFileContent) //row data
  80.     } catch {
  81.       case e: Exception => {
  82.         e.printStackTrace() //please use log
  83.       }
  84.     }
  85.     IOUtils.closeQuietly(client)
  86.   }
  89.   def toSubmit(user: String, code: String): JobExecuteResult = {
  90.     // 1. build  params
  91.     // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
  92.     val labels: util.Map[String, Any] = new util.HashMap[String, Any]
  93.     labels.put(LabelKeyConstant.ENGINE_TYPE_KEY, "spark-2.4.3"); // required engineType Label
  94.     labels.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, user + "-APPName"); // 请求的用户和应用名，两个参数都不能少，其中APPName不能带有"-"建议替换为"_"
  95.     labels.put(LabelKeyConstant.CODE_TYPE_KEY, "py"); // 指定脚本类型
  97.     val startupMap = new java.util.HashMap[String, Any]()
  98.     // Support setting engine native parameters,For example: parameters of engines such as spark/hive
  99.     startupMap.put("spark.executor.instances", 2);
  100.     // setting linkis params
  101.     startupMap.put("wds.linkis.rm.yarnqueue", "default");
  102.     // 2. build jobSubmitAction
  103.     val jobSubmitAction = JobSubmitAction.builder
  104.       .addExecuteCode(code)
  105.       .setStartupParams(startupMap)
  106.       .setUser(user) //submit user
  107.       .addExecuteUser(user) //execute user
  108.       .setLabels(labels)
  109.       .build
  110.     // 3. to execute
  111.     client.submit(jobSubmitAction)
  112.   }
  113. }