Data Source Client SDK

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. <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>

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

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("message queue")
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. }