influxdb-client-scala

The reference Scala client that allows query and write for the InfluxDB 2.x by Pekko Streams. The client is cross-built against Scala 2.12 and 2.13.

Documentation

This section contains links to the client library documentation.

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

Queries

The QueryScalaApi is based on the Pekko Streams.

The following example demonstrates querying using the Flux language:

1. package example

3. import org.apache.pekko.actor.ActorSystem
4. import org.apache.pekko.stream.scaladsl.Sink
5. import com.influxdb.client.scala.InfluxDBClientScalaFactory
6. import com.influxdb.query.FluxRecord

8. import scala.concurrent.Await
9. import scala.concurrent.duration.Duration

11. object InfluxDB2ScalaExample {

13. implicit val system: ActorSystem = ActorSystem("it-tests")

15. def main(args: Array[String]): Unit = {

17. val influxDBClient = InfluxDBClientScalaFactory
18. .create("http://localhost:8086", "my-token".toCharArray, "my-org")

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

24. //Result is returned as a stream
25. val results = influxDBClient.getQueryScalaApi().query(fluxQuery)

27. //Example of additional result stream processing on client side
28. val sink = results
29. //filter on client side using `filter` built-in operator
30. .filter(it => "cpu0" == it.getValueByKey("cpu"))
31. //take first 20 records
32. .take(20)
33. //print results
34. .runWith(Sink.foreach[FluxRecord](it => println(s"Measurement: ${it.getMeasurement}, value: ${it.getValue}")
35. ))

37. // wait to finish
38. Await.result(sink, Duration.Inf)

40. influxDBClient.close()
41. system.terminate()
42. }
43. }

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

1. package example

3. import org.apache.pekko.actor.ActorSystem
4. import org.apache.pekko.stream.scaladsl.Sink
5. import com.influxdb.client.scala.InfluxDBClientScalaFactory

7. import scala.concurrent.Await
8. import scala.concurrent.duration.Duration

10. object InfluxDB2ScalaExampleRaw {

12. implicit val system: ActorSystem = ActorSystem("it-tests")

14. def main(args: Array[String]): Unit = {
15. val influxDBClient = InfluxDBClientScalaFactory
16. .create("http://localhost:8086", "my-token".toCharArray, "my-org")

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

23. //Result is returned as a stream
24. val sink = influxDBClient.getQueryScalaApi().queryRaw(fluxQuery)
25. //print results
26. .runWith(Sink.foreach[String](it => println(s"Line: $it")))

28. // wait to finish
29. Await.result(sink, Duration.Inf)

31. influxDBClient.close()
32. system.terminate()
33. }
34. }

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

InfluxDBClientScala 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 java.time.temporal.ChronoUnit

5. import org.apache.pekko.actor.ActorSystem
6. import org.apache.pekko.stream.scaladsl.Sink
7. import com.influxdb.client.scala.InfluxDBClientScalaFactory
8. import com.influxdb.query.FluxRecord
9. import com.influxdb.query.dsl.Flux
10. import com.influxdb.query.dsl.functions.restriction.Restrictions

12. import scala.concurrent.Await
13. import scala.concurrent.duration.Duration

15. object InfluxDB2ScalaExampleDSL {

17. implicit val system: ActorSystem = ActorSystem("it-tests")

19. def main(args: Array[String]) {

21. val influxDBClient = InfluxDBClientScalaFactory
22. .create("http://localhost:8086", "my-token".toCharArray, "my-org")

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

28. //Result is returned as a stream
29. val results = influxDBClient.getQueryScalaApi().query(mem.toString())

31. //Example of additional result stream processing on client side
32. val sink = results
33. //filter on client side using `filter` built-in operator
34. .filter(it => it.getValue.asInstanceOf[Double] > 55)
35. //take first 20 records
36. .take(20)
37. //print results
38. .runWith(Sink.foreach[FluxRecord](it => println(s"Measurement: ${it.getMeasurement}, value: ${it.getValue}")))

40. // wait to finish
41. Await.result(sink, Duration.Inf)

43. influxDBClient.close()
44. system.terminate()
45. }
46. }

Version

Scala 2.12

The latest version for Maven dependency:

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

Or when using with Gradle:

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

Scala 2.13

The latest version for Maven dependency:

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

Or when using with Gradle:

1. dependencies {
2. implementation "com.influxdb:influxdb-client-scala_2.13: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. }