Use of 1.0 SDK

Linkis provides a convenient interface for calling JAVA and SCALA. You only need to introduce the linkis-computation-client module to use it. After 1.0, the method of submitting with Label is added. The following will be compatible with 0.X and 1.0. New way to introduce

1 Introduce dependent modules

  1. <dependency>
  2.   <groupId>com.webank.wedatasphere.linkis</groupId>
  3.   <artifactId>linkis-computation-client</artifactId>
  4.   <version>${linkis.version}</version>
  5. </dependency>
  6. like:
  7. <dependency>
  8.   <groupId>com.webank.wedatasphere.linkis</groupId>
  9.   <artifactId>linkis-computation-client</artifactId>
  10.   <version>1.0.0-RC1</version>
  11. </dependency>

2 Compatible with 0.X Execute method submission

2.1 Java test code

Establish the Java test class UJESClientImplTestJ, the specific interface meaning can be seen in the notes:

  1. package com.webank.wedatasphere.linkis.client.test
  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.authentication.TokenAuthenticationStrategy;
  7. import com.webank.wedatasphere.linkis.httpclient.dws.config.DWSClientConfig;
  8. import com.webank.wedatasphere.linkis.httpclient.dws.config.DWSClientConfigBuilder;
  9. import com.webank.wedatasphere.linkis.ujes.client.UJESClient;
  10. import com.webank.wedatasphere.linkis.ujes.client.UJESClientImpl;
  11. import com.webank.wedatasphere.linkis.ujes.client.request.JobExecuteAction;
  12. import com.webank.wedatasphere.linkis.ujes.client.request.ResultSetAction;
  13. import com.webank.wedatasphere.linkis.ujes.client.response.JobExecuteResult;
  14. import com.webank.wedatasphere.linkis.ujes.client.response.JobInfoResult;
  15. import com.webank.wedatasphere.linkis.ujes.client.response.JobProgressResult;
  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.     public static void main(String[] args){
  30.         String user = "hadoop";
  31.         String executeCode = "show databases;";
  34.         // 1. Configure DWSClientBuilder, get a DWSClientConfig through DWSClientBuilder
  35.         DWSClientConfig clientConfig = ((DWSClientConfigBuilder) (DWSClientConfigBuilder.newBuilder()
  36.                 .addServerUrl("http://${ip}:${port}") //Specify ServerUrl, the address of the linkis server-side gateway, such as http://{ip}:{port}
  37.                 .connectionTimeout(30000) //connectionTimeOut client connection timeout
  38.                 .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES) //Whether to enable registration discovery, if enabled, the newly launched Gateway will be automatically discovered
  39.                 .loadbalancerEnabled(true) // Whether to enable load balancing, if registration discovery is not enabled, load balancing is meaningless
  40.                 .maxConnectionSize(5) //Specify the maximum number of connections, that is, the maximum number of concurrent
  41.                 .retryEnabled(false).readTimeout(30000) //execution failed, whether to allow retry
  42.                 .setAuthenticationStrategy(new StaticAuthenticationStrategy()) //AuthenticationStrategy Linkis authentication method
  43.                 .setAuthTokenKey("${username}").setAuthTokenValue("${password}"))) //Authentication key, generally the user name; authentication value, generally the password corresponding to the user name
  44.                 .setDWSVersion("v1").build(); //Linkis background protocol version, the current version is v1
  47.         // 2. Get a UJESClient through DWSClientConfig
  48.         UJESClient client = new UJESClientImpl(clientConfig);
  51.         try {
  52.             // 3. Start code execution
  53.             System.out.println("user: "+ user + ", code: [" + executeCode + "]");
  54.             Map<String, Object> startupMap = new HashMap<String, Object>();
  55.             startupMap.put("wds.linkis.yarnqueue", "default"); // Various startup parameters can be stored in startupMap, see linkis management console configuration
  56.             JobExecuteResult jobExecuteResult = client.execute(JobExecuteAction.builder()
  57.                     .setCreator("linkisClient-Test") //creator, the system name of the client requesting linkis, used for system-level isolation
  58.                     .addExecuteCode(executeCode) //ExecutionCode The code to be executed
  59.                     .setEngineType((JobExecuteAction.EngineType) JobExecuteAction.EngineType$.MODULE$.HIVE()) // The execution engine type of the linkis that you want to request, such as Spark hive, etc.
  60.                     .setUser(user) //User, request user; used for user-level multi-tenant isolation
  61.                     .setStartupParams(startupMap)
  62.                     .build());
  63.             System.out.println("execId: "+ jobExecuteResult.getExecID() + ", taskId:" + jobExecuteResult.taskID());
  66.             // 4. Get the execution status of the script
  67.             JobInfoResult jobInfoResult = client.getJobInfo(jobExecuteResult);
  68.             int sleepTimeMills = 1000;
  69.             while(!jobInfoResult.isCompleted()) {
  70.                 // 5. Get the execution progress of the script
  71.                 JobProgressResult progress = client.progress(jobExecuteResult);
  72.                 Utils.sleepQuietly(sleepTimeMills);
  73.                 jobInfoResult = client.getJobInfo(jobExecuteResult);
  74.             }
  77.             // 6. Get the job information of the script
  78.             JobInfoResult jobInfo = client.getJobInfo(jobExecuteResult);
  79.             // 7. Get the list of result sets (if the user submits multiple SQL at a time, multiple result sets will be generated)
  80.             String resultSet = jobInfo.getResultSetList(client)[0];
  81.             // 8. Get a specific result set through a result set information
  82.                         Object fileContents = client.resultSet(ResultSetAction.builder().setPath(resultSet).setUser(jobExecuteResult.getUser()).build()).getFileContent();
  83.                         System.out.println("fileContents: "+ fileContents);
  85.                     } catch (Exception e) {
  86.                         e.printStackTrace();
  87.                         IOUtils.closeQuietly(client);
  88.                     }
  89.                     IOUtils.closeQuietly(client);
  90.                 }
  91.             }

Run the above code to interact with Linkis

2.2 Scala test code

  1. package com.webank.wedatasphere.linkis.client.test
  4. import java.util.concurrent.TimeUnit
  7. import com.webank.wedatasphere.linkis.common.utils.Utils
  8. import com.webank.wedatasphere.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy
  9. import com.webank.wedatasphere.linkis.httpclient.dws.config.DWSClientConfigBuilder
  10. import com.webank.wedatasphere.linkis.ujes.client.UJESClient
  11. import com.webank.wedatasphere.linkis.ujes.client.request.JobExecuteAction.EngineType
  12. import com.webank.wedatasphere.linkis.ujes.client.request.{JobExecuteAction, ResultSetAction}
  13. import org.apache.commons.io.IOUtils
  16. object LinkisClientImplTest extends App {
  19.   var executeCode = "show databases;"
  20.   var user = "hadoop"
  23.   // 1. Configure DWSClientBuilder, get a DWSClientConfig through DWSClientBuilder
  24.   val clientConfig = DWSClientConfigBuilder.newBuilder()
  25.     .addServerUrl("http://${ip}:${port}") //Specify ServerUrl, the address of the Linkis server-side gateway, such as http://{ip}:{port}
  26.     .connectionTimeout(30000) //connectionTimeOut client connection timeout
  27.     .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES) //Whether to enable registration discovery, if enabled, the newly launched Gateway will be automatically discovered
  28.     .loadbalancerEnabled(true) // Whether to enable load balancing, if registration discovery is not enabled, load balancing is meaningless
  29.     .maxConnectionSize(5) //Specify the maximum number of connections, that is, the maximum number of concurrent
  30.     .retryEnabled(false).readTimeout(30000) //execution failed, whether to allow retry
  31.     .setAuthenticationStrategy(new StaticAuthenticationStrategy()) //AuthenticationStrategy Linkis authentication method
  32.     .setAuthTokenKey("${username}").setAuthTokenValue("${password}") //Authentication key, generally the user name; authentication value, generally the password corresponding to the user name
  33.     .setDWSVersion("v1").build() //Linkis backend protocol version, the current version is v1
  36.   // 2. Get a UJESClient through DWSClientConfig
  37.   val client = UJESClient(clientConfig)
  40.   try {
  41.     // 3. Start code execution
  42.     println("user: "+ user + ", code: [" + executeCode + "]")
  43.     val startupMap = new java.util.HashMap[String, Any]()
  44.     startupMap.put("wds.linkis.yarnqueue", "default") //Startup parameter configuration
  45.     val jobExecuteResult = client.execute(JobExecuteAction.builder()
  46.       .setCreator("LinkisClient-Test") //creator, requesting the system name of the Linkis client, used for system-level isolation
  47.       .addExecuteCode(executeCode) //ExecutionCode The code to be executed
  48.       .setEngineType(EngineType.SPARK) // The execution engine type of Linkis that you want to request, such as Spark hive, etc.
  49.       .setStartupParams(startupMap)
  50.       .setUser(user).build()) //User, request user; used for user-level multi-tenant isolation
  51.     println("execId: "+ jobExecuteResult.getExecID + ", taskId:" + jobExecuteResult.taskID)
  54.     // 4. Get the execution status of the script
  55.     var jobInfoResult = client.getJobInfo(jobExecuteResult)
  56.     val sleepTimeMills: Int = 1000
  57.     while (!jobInfoResult.isCompleted) {
  58.       // 5. Get the execution progress of the script
  59.       val progress = client.progress(jobExecuteResult)
  60.       val progressInfo = if (progress.getProgressInfo != null) progress.getProgressInfo.toList else List.empty
  61.       println("progress: "+ progress.getProgress + ", progressInfo:" + progressInfo)
  62.       Utils.sleepQuietly(sleepTimeMills)
  63.       jobInfoResult = client.getJobInfo(jobExecuteResult)
  64.     }
  65.     if (!jobInfoResult.isSucceed) {
  66.       println("Failed to execute job: "+ jobInfoResult.getMessage)
  67.       throw new Exception(jobInfoResult.getMessage)
  68.     }
  71.     // 6. Get the job information of the script
  72.     val jobInfo = client.getJobInfo(jobExecuteResult)
  73.     // 7. Get the list of result sets (if the user submits multiple SQL at a time, multiple result sets will be generated)
  74.     val resultSetList = jobInfoResult.getResultSetList(client)
  75.     println("All result set list:")
  76.     resultSetList.foreach(println)
  77.     val oneResultSet = jobInfo.getResultSetList(client).head
  78.     // 8. Get a specific result set through a result set information
  79.     val fileContents = client.resultSet(ResultSetAction.builder().setPath(oneResultSet).setUser(jobExecuteResult.getUser).build()).getFileContent
  80.     println("First fileContents: ")
  81.     println(fileContents)
  82.   } catch {
  83.     case e: Exception => {
  84.       e.printStackTrace()
  85.     }
  86.   }
  87.   IOUtils.closeQuietly(client)
  88. }

3 1.0 New Submit Method

1.0 adds the client.submit method, which is used to interface with the new task execution interface of 1.0, and supports the input of Label and other parameters

3.1 Java Test Class

  1. package com.webank.wedatasphere.linkis.client.test;
  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.DWSClientConfig;
  7. import com.webank.wedatasphere.linkis.httpclient.dws.config.DWSClientConfigBuilder;
  8. import com.webank.wedatasphere.linkis.manager.label.constant.LabelKeyConstant;
  9. import com.webank.wedatasphere.linkis.protocol.constants.TaskConstant;
  10. import com.webank.wedatasphere.linkis.ujes.client.UJESClient;
  11. import com.webank.wedatasphere.linkis.ujes.client.UJESClientImpl;
  12. import com.webank.wedatasphere.linkis.ujes.client.request.JobSubmitAction;
  13. import com.webank.wedatasphere.linkis.ujes.client.request.ResultSetAction;
  14. import com.webank.wedatasphere.linkis.ujes.client.response.JobExecuteResult;
  15. import com.webank.wedatasphere.linkis.ujes.client.response.JobInfoResult;
  16. import com.webank.wedatasphere.linkis.ujes.client.response.JobProgressResult;
  17. import org.apache.commons.io.IOUtils;
  20. import java.util.HashMap;
  21. import java.util.Map;
  22. import java.util.concurrent.TimeUnit;
  25. public class JavaClientTest {
  28.     public static void main(String[] args){
  31.         String user = "hadoop";
  32.         String executeCode = "show tables";
  35.         // 1. Configure ClientBuilder and get ClientConfig
  36.         DWSClientConfig clientConfig = ((DWSClientConfigBuilder) (DWSClientConfigBuilder.newBuilder()
  37.                 .addServerUrl("http://${ip}:${port}") //Specify ServerUrl, the address of the linkis server-side gateway, such as http://{ip}:{port}
  38.                 .connectionTimeout(30000) //connectionTimeOut client connection timeout
  39.                 .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES) //Whether to enable registration discovery, if enabled, the newly launched Gateway will be automatically discovered
  40.                 .loadbalancerEnabled(true) // Whether to enable load balancing, if registration discovery is not enabled, load balancing is meaningless
  41.                 .maxConnectionSize(5) //Specify the maximum number of connections, that is, the maximum number of concurrent
  42.                 .retryEnabled(false).readTimeout(30000) //execution failed, whether to allow retry
  43.                 .setAuthenticationStrategy(new StaticAuthenticationStrategy()) //AuthenticationStrategy Linkis authentication method
  44.                 .setAuthTokenKey("${username}").setAuthTokenValue("${password}"))) //Authentication key, generally the user name; authentication value, generally the password corresponding to the user name
  45.                 .setDWSVersion("v1").build(); //Linkis background protocol version, the current version is v1
  48.         // 2. Get a UJESClient through DWSClientConfig
  49.         UJESClient client = new UJESClientImpl(clientConfig);
  52.         try {
  53.             // 3. Start code execution
  54.             System.out.println("user: "+ user + ", code: [" + executeCode + "]");
  55.             Map<String, Object> startupMap = new HashMap<String, Object>();
  56.             // Various startup parameters can be stored in startupMap, see linkis management console configuration
  57.             startupMap.put("wds.linkis.yarnqueue", "q02");
  58.             //Specify Label
  59.             Map<String, Object> labels = new HashMap<String, Object>();
  60.             //Add the label that this execution depends on: EngineTypeLabel/UserCreatorLabel/EngineRunTypeLabel
  61.             labels.put(LabelKeyConstant.ENGINE_TYPE_KEY, "hive-1.2.1");
  62.             labels.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, "hadoop-IDE");
  63.             labels.put(LabelKeyConstant.ENGINE_RUN_TYPE_KEY, "hql");
  64.             //Specify source
  65.             Map<String, Object> source = new HashMap<String, Object>();
  66.             source.put(TaskConstant.SCRIPTPATH, "LinkisClient-test");
  67.             JobExecuteResult jobExecuteResult = client.submit( JobSubmitAction.builder()
  68.                     .addExecuteCode(executeCode)
  69.                     .setStartupParams(startupMap)
  70.                     .setUser(user)//Job submit user
  71.                     .addExecuteUser(user)//The actual execution user
  72.                     .setLabels(labels)
  73.                     .setSource(source)
  74.                     .build()
  75.             );
  76.             System.out.println("execId: "+ jobExecuteResult.getExecID() + ", taskId:" + jobExecuteResult.taskID());
  79.             // 4. Get the execution status of the script
  80.             JobInfoResult jobInfoResult = client.getJobInfo(jobExecuteResult);
  81.             int sleepTimeMills = 1000;
  82.             while(!jobInfoResult.isCompleted()) {
  83.                 // 5. Get the execution progress of the script
  84.                 JobProgressResult progress = client.progress(jobExecuteResult);
  85.                 Utils.sleepQuietly(sleepTimeMills);
  86.                 jobInfoResult = client.getJobInfo(jobExecuteResult);
  87.             }
  90.             // 6. Get the job information of the script
  91.             JobInfoResult jobInfo = client.getJobInfo(jobExecuteResult);
  92.             // 7. Get the list of result sets (if the user submits multiple SQL at a time, multiple result sets will be generated)
  93.             String resultSet = jobInfo.getResultSetList(client)[0];
  94.             // 8. Get a specific result set through a result set information
  95.             Object fileContents = client.resultSet(ResultSetAction.builder().setPath(resultSet).setUser(jobExecuteResult.getUser()).build()).getFileContent();
  96.             System.out.println("fileContents: "+ fileContents);
  99.         } catch (Exception e) {
  100.             e.printStackTrace();
  101.             IOUtils.closeQuietly(client);
  102.         }
  103.         IOUtils.closeQuietly(client);
  104.     }
  105. }

3.2 Scala Test Class

  1. package com.webank.wedatasphere.linkis.client.test
  4. import java.util
  5. import java.util.concurrent.TimeUnit
  8. import com.webank.wedatasphere.linkis.common.utils.Utils
  9. import com.webank.wedatasphere.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy
  10. import com.webank.wedatasphere.linkis.httpclient.dws.config.DWSClientConfigBuilder
  11. import com.webank.wedatasphere.linkis.manager.label.constant.LabelKeyConstant
  12. import com.webank.wedatasphere.linkis.protocol.constants.TaskConstant
  13. import com.webank.wedatasphere.linkis.ujes.client.UJESClient
  14. import com.webank.wedatasphere.linkis.ujes.client.request.{JobSubmitAction, ResultSetAction}
  15. import org.apache.commons.io.IOUtils
  20. object ScalaClientTest {
  23.   def main(args: Array[String]): Unit = {
  24.     val executeCode = "show tables"
  25.     val user = "hadoop"
  28.     // 1. Configure DWSClientBuilder, get a DWSClientConfig through DWSClientBuilder
  29.     val clientConfig = DWSClientConfigBuilder.newBuilder()
  30.       .addServerUrl("http://${ip}:${port}") //Specify ServerUrl, the address of the Linkis server-side gateway, such as http://{ip}:{port}
  31.       .connectionTimeout(30000) //connectionTimeOut client connection timeout
  32.       .discoveryEnabled(false).discoveryFrequency(1, TimeUnit.MINUTES) //Whether to enable registration discovery, if enabled, the newly launched Gateway will be automatically discovered
  33.       .loadbalancerEnabled(true) // Whether to enable load balancing, if registration discovery is not enabled, load balancing is meaningless
  34.       .maxConnectionSize(5) //Specify the maximum number of connections, that is, the maximum number of concurrent
  35.       .retryEnabled(false).readTimeout(30000) //execution failed, whether to allow retry
  36.       .setAuthenticationStrategy(new StaticAuthenticationStrategy()) //AuthenticationStrategy Linkis authentication method
  37.       .setAuthTokenKey("${username}").setAuthTokenValue("${password}") //Authentication key, generally the user name; authentication value, generally the password corresponding to the user name
  38.       .setDWSVersion("v1").build() //Linkis backend protocol version, the current version is v1
  41.     // 2. Get a UJESClient through DWSClientConfig
  42.     val client = UJESClient(clientConfig)
  45.     try {
  46.       // 3. Start code execution
  47.       println("user: "+ user + ", code: [" + executeCode + "]")
  48.       val startupMap = new java.util.HashMap[String, Any]()
  49.       startupMap.put("wds.linkis.yarnqueue", "q02") //Startup parameter configuration
  50.       //Specify Label
  51.       val labels: util.Map[String, Any] = new util.HashMap[String, Any]
  52.       //Add the label that this execution depends on, such as engineLabel
  53.       labels.put(LabelKeyConstant.ENGINE_TYPE_KEY, "hive-1.2.1")
  54.       labels.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, "hadoop-IDE")
  55.       labels.put(LabelKeyConstant.ENGINE_RUN_TYPE_KEY, "hql")
  56.       //Specify source
  57.       val source: util.Map[String, Any] = new util.HashMap[String, Any]
  58.       source.put(TaskConstant.SCRIPTPATH, "LinkisClient-test")
  59.       val jobExecuteResult = client.submit(JobSubmitAction.builder
  60.           .addExecuteCode(executeCode)
  61.           .setStartupParams(startupMap)
  62.           .setUser(user) //Job submit user
  63.           .addExecuteUser(user) //The actual execution user
  64.           .setLabels(labels)
  65.           .setSource(source)
  66.           .build) //User, requesting user; used for user-level multi-tenant isolation
  67.       println("execId: "+ jobExecuteResult.getExecID + ", taskId:" + jobExecuteResult.taskID)
  70.       // 4. Get the execution status of the script
  71.       var jobInfoResult = client.getJobInfo(jobExecuteResult)
  72.       val sleepTimeMills: Int = 1000
  73.       while (!jobInfoResult.isCompleted) {
  74.         // 5. Get the execution progress of the script
  75.         val progress = client.progress(jobExecuteResult)
  76.         val progressInfo = if (progress.getProgressInfo != null) progress.getProgressInfo.toList else List.empty
  77.         println("progress: "+ progress.getProgress + ", progressInfo:" + progressInfo)
  78.         Utils.sleepQuietly(sleepTimeMills)
  79.         jobInfoResult = client.getJobInfo(jobExecuteResult)
  80.       }
  81.       if (!jobInfoResult.isSucceed) {
  82.         println("Failed to execute job: "+ jobInfoResult.getMessage)
  83.         throw new Exception(jobInfoResult.getMessage)
  84.       }
  87.       // 6. Get the job information of the script
  88.       val jobInfo = client.getJobInfo(jobExecuteResult)
  89.       // 7. Get the list of result sets (if the user submits multiple SQL at a time, multiple result sets will be generated)
  90.       val resultSetList = jobInfoResult.getResultSetList(client)
  91.       println("All result set list:")
  92.       resultSetList.foreach(println)
  93.       val oneResultSet = jobInfo.getResultSetList(client).head
  94.       // 8. Get a specific result set through a result set information
  95.       val fileContents = client.resultSet(ResultSetAction.builder().setPath(oneResultSet).setUser(jobExecuteResult.getUser).build()).getFileContent
  96.       println("First fileContents: ")
  97.       println(fileContents)
  98.     } catch {
  99.       case e: Exception => {
  100.         e.printStackTrace()
  101.       }
  102.     }
  103.     IOUtils.closeQuietly(client)
  104.   }
  107. }