influxdb-client-kotlin

The reference Kotlin client that allows query and write for the InfluxDB 2.x by Kotlin Channel coroutines.

Documentation

This section contains links to the client library documentation.

• Product documentation, Getting Started
• Examples
• API Reference
• Changelog

Features

• Querying data using Flux language
• Writing data
• Advanced Usage

Queries

The QueryKotlinApi supports asynchronous queries by Kotlin Channel coroutines.

The following example demonstrates querying using the Flux language:

1. package example

3. import com.influxdb.client.kotlin.InfluxDBClientKotlinFactory
4. import kotlinx.coroutines.channels.consumeEach
5. import kotlinx.coroutines.channels.filter
6. import kotlinx.coroutines.channels.take
7. import kotlinx.coroutines.runBlocking

9. fun main(args: Array<String>) = runBlocking {

11. val influxDBClient = InfluxDBClientKotlinFactory
12. .create("http://localhost:8086", "my-token".toCharArray(), "my-org")

14. val fluxQuery = ("from(bucket: \"my-bucket\")\n"
15. + " |> range(start: -1d)"
16. + " |> filter(fn: (r) => (r[\"_measurement\"] == \"cpu\" and r[\"_field\"] == \"usage_system\"))")

18. //Result is returned as a stream
19. val results = influxDBClient.getQueryKotlinApi().query(fluxQuery)

21. //Example of additional result stream processing on client side
22. results
23. //filter on client side using `filter` built-in operator
24. .filter { "cpu0" == it.getValueByKey("cpu") }
25. //take first 20 records
26. .take(20)
27. //print results
28. .consumeEach { println("Measurement: ${it.measurement}, value: ${it.value}") }

30. influxDBClient.close()
31. }

It is possible to parse a result line-by-line using the queryRaw method:

1. package example

3. import com.influxdb.client.kotlin.InfluxDBClientKotlinFactory
4. import kotlinx.coroutines.channels.consumeEach
5. import kotlinx.coroutines.runBlocking

7. fun main(args: Array<String>) = runBlocking {

9. val influxDBClient = InfluxDBClientKotlinFactory
10. .create("http://localhost:8086", "my-token".toCharArray(), "my-org")

12. val fluxQuery = ("from(bucket: \"my-bucket\")\n"
13. + " |> range(start: -5m)"
14. + " |> filter(fn: (r) => (r[\"_measurement\"] == \"cpu\" and r[\"_field\"] == \"usage_system\"))"
15. + " |> sample(n: 5, pos: 1)")

17. //Result is returned as a stream
18. val results = influxDBClient.getQueryKotlinApi().queryRaw(fluxQuery)

20. //print results
21. results.consumeEach { println("Line: $it") }

23. influxDBClient.close()
24. }

Writes

The WriteKotlinApi supports ingest data by:

• DataPoint
• LineProtocol
• Data class
• List of above items

The following example shows how to use various type of data:

1. package example

3. import com.influxdb.annotations.Column
4. import com.influxdb.annotations.Measurement
5. import com.influxdb.client.domain.WritePrecision
6. import com.influxdb.client.kotlin.InfluxDBClientKotlinFactory
7. import com.influxdb.client.write.Point
8. import kotlinx.coroutines.flow.collect
9. import kotlinx.coroutines.flow.consumeAsFlow
10. import kotlinx.coroutines.runBlocking
11. import java.time.Instant

13. fun main() = runBlocking {

15. val org = "my-org"
16. val bucket = "my-bucket"

18. //
19. // Initialize client
20. //
21. val client = InfluxDBClientKotlinFactory
22. .create("http://localhost:8086", "my-token".toCharArray(), org, bucket)

24. val writeApi = client.getWriteKotlinApi()

26. //
27. // Write by Data Point
28. //
29. val point = Point.measurement("temperature")
30. .addTag("location", "west")
31. .addField("value", 55.0)
32. .time(Instant.now().toEpochMilli(), WritePrecision.MS)

34. writeApi.writePoint(point)

36. //
37. // Write by LineProtocol
38. //
39. writeApi.writeRecord("temperature,location=north value=60.0", WritePrecision.NS)

41. //
42. // Write by DataClass
43. //
44. val temperature = Temperature("south", 62.0, Instant.now())

46. writeApi.writeMeasurement(temperature, WritePrecision.NS)

48. //
49. // Query results
50. //
51. val fluxQuery =
52. """from(bucket: "$bucket") |> range(start: 0) |> filter(fn: (r) => (r["_measurement"] == "temperature"))"""

54. client
55. .getQueryKotlinApi()
56. .query(fluxQuery)
57. .consumeAsFlow()
58. .collect { println("Measurement: ${it.measurement}, value: ${it.value}") }

60. client.close()
61. }

63. @Measurement(name = "temperature")
64. data class Temperature(
65. @Column(tag = true) val location: String,
66. @Column val value: Double,
67. @Column(timestamp = true) val time: Instant
68. )

• sources - KotlinWriteApi.kt

Advanced Usage

Client configuration file

A client can be configured via configuration file. The configuration file has to be named as influx2.properties and has to be in root of classpath.

The following options are supported:

The influx2.readTimeout, influx2.writeTimeout and influx2.connectTimeout supports ms, s and m as unit. Default is milliseconds.

Configuration example

1. influx2.url=http://localhost:8086
2. influx2.org=my-org
3. influx2.bucket=my-bucket
4. influx2.token=my-token
5. influx2.logLevel=BODY
6. influx2.readTimeout=5s
7. influx2.writeTimeout=10s
8. influx2.connectTimeout=5s

and then:

1. val influxDBClient = InfluxDBClientKotlinFactory.create();

Client connection string

A client can be constructed using a connection string that can contain the InfluxDBClientOptions parameters encoded into the URL.

1. val influxDBClient = InfluxDBClientKotlinFactory
2. .create("http://localhost:8086?readTimeout=5000&connectTimeout=5000&logLevel=BASIC", token)

The following options are supported:

The readTimeout, writeTimeout and connectTimeout supports ms, s and m as unit. Default is milliseconds.

Gzip support

InfluxDBClientKotlin does not enable gzip compress for http requests by default. If you want to enable gzip to reduce transfer data’s size, you can call:

1. influxDBClient.enableGzip();

Log HTTP Request and Response

The Requests and Responses can be logged by changing the LogLevel. LogLevel values are NONE, BASIC, HEADER, BODY. Note that applying the BODY LogLevel will disable chunking while streaming and will load the whole response into memory.

1. influxDBClient.setLogLevel(LogLevel.HEADERS)

Check the server status

Server availability can be checked using the influxDBClient.ping() endpoint.

Construct queries using the flux-dsl query builder

1. package example

3. import com.influxdb.client.kotlin.InfluxDBClientKotlinFactory
4. import com.influxdb.query.dsl.Flux
5. import com.influxdb.query.dsl.functions.restriction.Restrictions
6. import kotlinx.coroutines.channels.consumeEach
7. import kotlinx.coroutines.channels.filter
8. import kotlinx.coroutines.channels.take
9. import kotlinx.coroutines.runBlocking
10. import java.time.temporal.ChronoUnit

12. fun main(args: Array<String>) = runBlocking {

14. val influxDBClient = InfluxDBClientKotlinFactory
15. .create("http://localhost:8086", "my-token".toCharArray(), "my-org")

17. val mem = Flux.from("my-bucket")
18. .range(-30L, ChronoUnit.MINUTES)
19. .filter(Restrictions.and(Restrictions.measurement().equal("mem"), Restrictions.field().equal("used_percent")))

21. //Result is returned as a stream
22. val results = influxDBClient.getQueryKotlinApi().query(mem.toString())

24. //Example of additional result stream processing on client side
25. results
26. //filter on client side using `filter` built-in operator
27. .filter { (it.value as Double) > 55 }
28. // take first 20 records
29. .take(20)
30. //print results
31. .consumeEach { println("Measurement: ${it.measurement}, value: ${it.value}") }

33. influxDBClient.close()
34. }

Version

The latest version for Maven dependency:

1. <dependency>
2. <groupId>com.influxdb</groupId>
3. <artifactId>influxdb-client-kotlin</artifactId>
4. <version>7.2.0</version>
5. </dependency>

Or when using with Gradle:

1. dependencies {
2. implementation "com.influxdb:influxdb-client-kotlin:7.2.0"
3. }

Snapshot Repository

The snapshots are deployed into OSS Snapshot repository.

Maven

1. <repository>
2. <id>ossrh</id>
3. <name>OSS Snapshot repository</name>
4. <url>https://oss.sonatype.org/content/repositories/snapshots/</url>
5. <releases>
6. <enabled>false</enabled>
7. </releases>
8. <snapshots>
9. <enabled>true</enabled>
10. </snapshots>
11. </repository>

Gradle

1. repositories {
2. maven { url "https://oss.sonatype.org/content/repositories/snapshots" }
3. }