JAVA SDK Manual

Linkis provides a convenient interface for calling JAVA and SCALA. It can be used only by introducing the linkis-computation-client module. After 1.0, the method of submitting with Label is added. The following will introduce both ways that compatible with 0.X and newly added in 1.0.

1. Introduce dependent modules

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

2. Java test code

Create the Java test class LinkisClientTest. Refer to the comments to understand the purposes of those interfaces:

  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("BML-AUTH"))
  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. build job and execute
  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 + "-IDE");// required execute user and creator
  104.         labels.put(LabelKeyConstant.CODE_TYPE_KEY, "py"); // required codeType
  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("IDE")  //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. }

Run the above code to interact with Linkis

3. Scala test code

Create the Scala test class LinkisClientTest. Refer to the comments to understand the purposes of those interfaces:

  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.       val jobExecuteResult = toSubmit(user, executeCode)
  50.       //0.X: val jobExecuteResult = toExecute(user, executeCode)
  51.       println("execId: " + jobExecuteResult.getExecID + ", taskId: " + jobExecuteResult.taskID)
  52.       // 4. get job jonfo
  53.       var jobInfoResult = client.getJobInfo(jobExecuteResult)
  54.       var logFromLen = 0
  55.       val logSize = 100
  56.       val sleepTimeMills : Int = 1000
  57.       while (!jobInfoResult.isCompleted) {
  58.         // 5. get progress and log
  59.         val progress = client.progress(jobExecuteResult)
  60.        println("progress: " + progress.getProgress)
  61.         val logObj = client .log(jobExecuteResult, logFromLen, logSize)
  62.         logFromLen = logObj.fromLine
  63.         val logArray = logObj.getLog
  64.         // 0: info 1: warn 2: error 3: all
  65.         if (logArray != null && logArray.size >= 4 && StringUtils.isNotEmpty(logArray.get(3))) {
  66.           println(s"log: ${logArray.get(3)}")
  67.         }
  68.         Utils.sleepQuietly(sleepTimeMills)
  69.         jobInfoResult = client.getJobInfo(jobExecuteResult)
  70.       }
  71.       if (!jobInfoResult.isSucceed) {
  72.         println("Failed to execute job: " + jobInfoResult.getMessage)
  73.         throw new Exception(jobInfoResult.getMessage)
  74.       }
  77.       // 6. Get the result set list (if the user submits multiple SQLs at a time,
  78.       // multiple result sets will be generated)
  79.       val jobInfo = client.getJobInfo(jobExecuteResult)
  80.       val resultSetList = jobInfoResult.getResultSetList(client)
  81.       println("All result set list:")
  82.       resultSetList.foreach(println)
  83.       val oneResultSet = jobInfo.getResultSetList(client).head
  84.       // 7. get resultContent
  85.       val fileContents = client.resultSet(ResultSetAction.builder().setPath(oneResultSet).setUser(jobExecuteResult.getUser).build()).getFileContent
  86.       println("First fileContents: ")
  87.       println(fileContents)
  88.     } catch {
  89.       case e: Exception => {
  90.         e.printStackTrace()
  91.       }
  92.     }
  93.     IOUtils.closeQuietly(client)
  94.   }
  97.   /**
  98.    * Linkis 1.0 recommends the use of Submit method
  99.    */
  100.   def toSubmit(user: String, code: String): JobExecuteResult = {
  101.     // 1. build  params
  102.     // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
  103.     val labels: util.Map[String, Any] = new util.HashMap[String, Any]
  104.     labels.put(LabelKeyConstant.ENGINE_TYPE_KEY, "spark-2.4.3"); // required engineType Label
  105.     labels.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, user + "-IDE");// required execute user and creator
  106.     labels.put(LabelKeyConstant.CODE_TYPE_KEY, "py"); // required codeType
  109.     val startupMap = new java.util.HashMap[String, Any]()
  110.     // Support setting engine native parameters,For example: parameters of engines such as spark/hive
  111.     startupMap.put("spark.executor.instances", 2);
  112.     // setting linkis params
  113.     startupMap.put("wds.linkis.rm.yarnqueue", "dws");
  114.     // 2. build jobSubmitAction
  115.     val jobSubmitAction = JobSubmitAction.builder
  116.       .addExecuteCode(code)
  117.       .setStartupParams(startupMap)
  118.       .setUser(user) //submit user
  119.       .addExecuteUser(user) //execute user
  120.       .setLabels(labels)
  121.       .build
  122.     // 3. to execute
  123.     client.submit(jobSubmitAction)
  124.   }
  129.   /**
  130.    * Compatible with 0.X execution mode
  131.    */
  132.   def toExecute(user: String, code: String): JobExecuteResult = {
  133.     // 1. build  params
  134.     // set label map :EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel/Tenant
  135.     val labels = new util.HashMap[String, Any]
  136.     // labels.put(LabelKeyConstant.TENANT_KEY, "fate");
  139.     val startupMap = new java.util.HashMap[String, Any]()
  140.     // Support setting engine native parameters,For example: parameters of engines such as spark/hive
  141.     startupMap.put("spark.executor.instances", 2)
  142.     // setting linkis params
  143.     startupMap.put("wds.linkis.rm.yarnqueue", "dws")
  144.     // 2. build JobExecuteAction (0.X old way of using)
  145.     val  executionAction = JobExecuteAction.builder()
  146.       .setCreator("IDE")  //creator, the system name of the client requesting linkis, used for system-level isolation
  147.       .addExecuteCode(code)   //Execution Code
  148.       .setEngineTypeStr("spark") // engineConn type
  149.       .setRunTypeStr("py") // code type
  150.       .setUser(user)   //execute user
  151.       .setStartupParams(startupMap) // start up params
  152.       .build();
  153.     executionAction.addRequestPayload(TaskConstant.LABELS, labels);
  154.     // 3. to execute
  155.     client.execute(executionAction)
  156.   }
  161. }