数据源 Client SDK 的使用

Linkis DataSource 提供了方便的JAVA和SCALA调用的接口，只需要引入linkis-datasource-client的模块就可以进行使用，

  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>

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

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