DataSource Client SDK

  1. Linkis DataSource provides a convenient interface for JAVA and SCALA calls, which can be used only by introducing the module of linkis-datasource-client

1. import dependent modules

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

2. Scala Test Code

Create a Scala test class LinkisDataSourceClientTest, the specific interface meaning can be seen in the comments：

  3. import com.fasterxml.jackson.databind.ObjectMapper
  4. import org.apache.linkis.common.utils.JsonUtils
  5. import org.apache.linkis.datasource.client.impl.{LinkisDataSourceRemoteClient, LinkisMetaDataRemoteClient}
  6. import org.apache.linkis.datasource.client.request.{CreateDataSourceAction, GetAllDataSourceTypesAction, MetadataGetDatabasesAction, UpdateDataSourceParameterAction}
  7. import org.apache.linkis.datasourcemanager.common.domain.{DataSource, DataSourceType}
  8. import org.apache.linkis.httpclient.dws.authentication.StaticAuthenticationStrategy
  9. import org.apache.linkis.httpclient.dws.config.{DWSClientConfig, DWSClientConfigBuilder}
  12. import java.io.StringWriter
  13. import java.util
  14. import java.util.concurrent.TimeUnit
  19. object LinkisDataSourceClientTest {
  22.   def main(args: Array[String]): Unit = {
  23.     val clientConfig =DWSClientConfigBuilder.newBuilder
  24.       .addServerUrl("http://127.0.0.1:9001") //set linkis-mg-gateway url: http://{ip}:{port}
  25.       .connectionTimeout(30000) //connection timtout
  26.       .discoveryEnabled(false) //disable discovery
  27.       .discoveryFrequency(1, TimeUnit.MINUTES)  // discovery frequency
  28.       .loadbalancerEnabled(true) // enable loadbalance
  29.       .maxConnectionSize(5) // set max Connection
  30.       .retryEnabled(false) // set retry
  31.       .readTimeout(30000) //set read timeout
  32.       .setAuthenticationStrategy(new StaticAuthenticationStrategy) //AuthenticationStrategy Linkis authen suppory static and Token
  33.       .setAuthTokenKey("hadoop")  // set submit user
  34.       .setAuthTokenValue("xxx") // set passwd or token
  35.       .setDWSVersion("v1") //linkis rest version v1
  36.       .build
  37.      //init datasource remote client   
  38.      val dataSourceClient = new LinkisDataSourceRemoteClient(clientConfig)
  39.      //init metadata remote client
  40.      val metaDataClient = new LinkisMetaDataRemoteClient(clientConfig)
  41.     //get all datasource type
  42.     testGetAllDataSourceTypes(dataSourceClient)
  43.     //create kafka datasource
  44.     testCreateDataSourceForKafka(dataSourceClient)
  45.     //create es datasource
  46.     testCreateDataSourceForEs(dataSourceClient)
  47.     //update datasource parameter for kafka
  48.     testUpdateDataSourceParameterForKafka(dataSourceClient)
  49.     //update datasource parameter for es
  50.     testUpdateDataSourceParameterForEs(dataSourceClient)
  51.     //get hive metadata database list
  52.     testMetadataGetDatabases(metaDataClient)
  53.   }
  56. def testGetAllDataSourceTypes(client:LinkisDataSourceRemoteClient): Unit ={
  57.     val getAllDataSourceTypesResult = client.getAllDataSourceTypes(GetAllDataSourceTypesAction.builder().setUser("hadoop").build()).getAllDataSourceType
  58.     System.out.println(getAllDataSourceTypesResult)
  59.   }
  62. def testCreateDataSourceForKafka(client:LinkisDataSourceRemoteClient): Unit ={
  63.     val dataSource = new DataSource();
  64.     val dataSourceType = new DataSourceType
  65.     dataSourceType.setName("kafka")
  66.     dataSourceType.setId("2")
  67.     dataSourceType.setLayers(2)
  68.     dataSourceType.setClassifier("message queue")
  69.     dataSourceType.setDescription("kafka")
  70.     dataSource.setDataSourceType(dataSourceType)
  71.     dataSource.setDataSourceName("kafka-test")
  72.     dataSource.setCreateSystem("client")
  73.     dataSource.setDataSourceTypeId(2l);
  76.     val dsJsonWriter = new StringWriter
  79.     val mapper = new ObjectMapper
  80.     JsonUtils.jackson.writeValue(dsJsonWriter, dataSource)
  81.     val map = mapper.readValue(dsJsonWriter.toString,new util.HashMap[String,Any]().getClass)
  82.     val id = client.createDataSource(CreateDataSourceAction.builder().setUser("hadoop").addRequestPayloads(map).build()).getInsert_id
  83.     System.out.println(id)
  86.   }
  89. def testCreateDataSourceForEs(client:LinkisDataSourceRemoteClient): Unit ={
  90.     val dataSource = new DataSource();
  91.     dataSource.setDataSourceName("es-test")
  92.     dataSource.setCreateSystem("client")
  93.     dataSource.setDataSourceTypeId(7l);
  94.     val dsJsonWriter = new StringWriter
  95.     val mapper = new ObjectMapper
  96.     JsonUtils.jackson.writeValue(dsJsonWriter, dataSource)
  97.     val map = mapper.readValue(dsJsonWriter.toString,new util.HashMap[String,Any]().getClass)
  98.     val id = client.createDataSource(CreateDataSourceAction.builder().setUser("hadoop").addRequestPayloads(map).build()).getInsert_id
  99.     System.out.println(id)
  102.   }
  105. def testUpdateDataSourceParameterForKafka(client:LinkisDataSourceRemoteClient): Unit ={
  106.     val params = new util.HashMap[String,Any]()
  107.     val connParams = new util.HashMap[String,Any]()
  108.     connParams.put("brokers","172.24.2.232:9092")
  109.     params.put("connectParams",connParams)
  110.     params.put("comment","kafka data source")
  111.   client.updateDataSourceParameter(UpdateDataSourceParameterAction.builder().setUser("hadoop").setDataSourceId("7").addRequestPayloads(params).build())
  112.   }
  115. def testUpdateDataSourceParameterForEs(client:LinkisDataSourceRemoteClient): Unit ={
  116.     val params = new util.HashMap[String,Any]()
  117.     val connParams = new util.HashMap[String,Any]()
  118.     val elasticUrls = new util.ArrayList[String]()
  119.     elasticUrls.add("http://172.24.2.231:9200")
  120.     connParams.put("elasticUrls",elasticUrls)
  121.     params.put("connectParams",connParams)
  122.     params.put("comment","es data source")
  123.     client.updateDataSourceParameter(UpdateDataSourceParameterAction.builder().setUser("hadoop").setDataSourceId("8").addRequestPayloads(params).build())
  124.   }
  129.  def testMetadataGetDatabases(client:LinkisMetaDataRemoteClient): Unit ={
  130.   client.getDatabases(MetadataGetDatabasesAction.builder().setUser("hadoop").setDataSourceId(9l).setUser("hadoop").setSystem("client").build()).getDbs
  131.   }
  132. }