InfluxDB
The Alpakka InfluxDb connector provides Akka Streams integration for InfluxDB.
For more information about InfluxDB, please visit the InfluxDB Documentation
Project Info: Alpakka InfluxDB | |
---|---|
Artifact | com.lightbend.akka
akka-stream-alpakka-influxdb
1.1.2
|
JDK versions | OpenJDK 8 |
Scala versions | 2.12.7, 2.13.0 |
JPMS module name | akka.stream.alpakka.influxdb |
License | |
Readiness level |
Since 1.1.0, 2019-07-03
|
Home page | https://doc.akka.io/docs/alpakka/current |
API documentation | |
Forums | |
Release notes | In the documentation |
Issues | Github issues |
Sources | https://github.com/akka/alpakka |
Alpakka InfluxDB was added in Alpakka 1.1.0 in July 2019 and is marked as “API may change”. Please try it out and suggest improvements.
Furthermore, the major InfluxDB update to version 2.0 is expected to bring API and dependency changes to Alpakka InfluxDB.
Artifacts
- sbt
libraryDependencies += "com.lightbend.akka" %% "akka-stream-alpakka-influxdb" % "1.1.2"
- Maven
<dependency> <groupId>com.lightbend.akka</groupId> <artifactId>akka-stream-alpakka-influxdb_2.12</artifactId> <version>1.1.2</version> </dependency>
- Gradle
dependencies { compile group: 'com.lightbend.akka', name: 'akka-stream-alpakka-influxdb_2.12', version: '1.1.2' }
The table below shows direct dependencies of this module and the second tab shows all libraries it depends on transitively.
- Direct dependencies
Organization Artifact Version License com.typesafe.akka akka-stream_2.12 2.5.23 Apache License, Version 2.0 org.influxdb influxdb-java 2.15 The MIT License (MIT) org.scala-lang scala-library 2.12.7 BSD 3-Clause - Dependency tree
com.typesafe.akka akka-stream_2.12 2.5.23 Apache License, Version 2.0 com.typesafe.akka akka-actor_2.12 2.5.23 Apache License, Version 2.0 com.typesafe config 1.3.3 Apache License, Version 2.0 org.scala-lang.modules scala-java8-compat_2.12 0.8.0 BSD 3-clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause com.typesafe.akka akka-protobuf_2.12 2.5.23 Apache License, Version 2.0 org.scala-lang scala-library 2.12.7 BSD 3-Clause com.typesafe ssl-config-core_2.12 0.3.7 Apache-2.0 com.typesafe config 1.3.3 Apache License, Version 2.0 org.scala-lang.modules scala-parser-combinators_2.12 1.1.1 BSD 3-clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.reactivestreams reactive-streams 1.0.2 CC0 org.scala-lang scala-library 2.12.7 BSD 3-Clause org.influxdb influxdb-java 2.15 The MIT License (MIT) com.squareup.okhttp3 logging-interceptor 3.13.1 Apache 2.0 com.squareup.okhttp3 okhttp 3.13.1 Apache 2.0 com.squareup.okio okio 1.17.2 Apache 2.0 com.squareup.okhttp3 okhttp 3.13.1 Apache 2.0 com.squareup.okio okio 1.17.2 Apache 2.0 com.squareup.retrofit2 converter-moshi 2.5.0 Apache 2.0 com.squareup.moshi moshi 1.5.0 Apache 2.0 com.squareup.okio okio 1.17.2 Apache 2.0 com.squareup.retrofit2 retrofit 2.5.0 Apache 2.0 com.squareup.okhttp3 okhttp 3.13.1 Apache 2.0 com.squareup.okio okio 1.17.2 Apache 2.0 com.squareup.retrofit2 retrofit 2.5.0 Apache 2.0 com.squareup.okhttp3 okhttp 3.13.1 Apache 2.0 com.squareup.okio okio 1.17.2 Apache 2.0 org.msgpack msgpack-core 0.8.16 Apache 2 org.scala-lang scala-library 2.12.7 BSD 3-Clause
Set up InfluxDB client
Sources, Flows and Sinks provided by this connector need a prepared org.influxdb.InfluxDB
to access to InfluxDB.
- Scala
-
influxDB = InfluxDBFactory.connect(INFLUXDB_URL, USERNAME, PASSWORD); influxDB.setDatabase(DatabaseName); influxDB.query(new Query("CREATE DATABASE " + DatabaseName, DatabaseName));
- Java
-
final InfluxDB influxDB = InfluxDBFactory.connect(INFLUXDB_URL, USERNAME, PASSWORD); influxDB.setDatabase(databaseName); influxDB.query(new Query("CREATE DATABASE " + databaseName, databaseName)); return influxDB;
InfluxDB as Source and Sink
Now we can stream messages from or to InfluxDB by providing the InfluxDB
to the InfluxDbSource
InfluxDbSource
or the InfluxDbSink
. InfluxDbSink
.
- Scala
-
@Measurement(name = "cpu", database = "InfluxDbSpec") public class InfluxDbSpecCpu extends Cpu { public InfluxDbSpecCpu() { } public InfluxDbSpecCpu(Instant time, String hostname, String region, Double idle, Boolean happydevop, Long uptimeSecs) { super(time, hostname, region, idle, happydevop, uptimeSecs); } public InfluxDbSpecCpu cloneAt(Instant time) { return new InfluxDbSpecCpu(time, getHostname(), getRegion(), getIdle(), getHappydevop(), getUptimeSecs()); } }
- Java
-
@Measurement(name = "cpu", database = "InfluxDbTest") public class InfluxDbCpu extends Cpu { public InfluxDbCpu() {} public InfluxDbCpu( Instant time, String hostname, String region, Double idle, Boolean happydevop, Long uptimeSecs) { super(time, hostname, region, idle, happydevop, uptimeSecs); } public InfluxDbCpu cloneAt(Instant time) { return new InfluxDbCpu( time, getHostname(), getRegion(), getIdle(), getHappydevop(), getUptimeSecs()); } }
With typed source
Use InfluxDbSource.typed
and InfluxDbSink.typed
to create source and sink. The data is converted by InfluxDBMapper. The data is converted by InfluxDBMapper.
- Scala
-
val f1 = InfluxDbSource .typed(classOf[InfluxDbSpecCpu], InfluxDbReadSettings(), influxDB, query) .map { cpu: InfluxDbSpecCpu => { val clonedCpu = cpu.cloneAt(cpu.getTime.plusSeconds(60000)) List(InfluxDbWriteMessage(clonedCpu)) } } .runWith(InfluxDbSink.typed(classOf[InfluxDbSpecCpu]))
- Java
-
CompletionStage<Done> completionStage = InfluxDbSource.typed(InfluxDbCpu.class, InfluxDbReadSettings.Default(), influxDB, query) .map( cpu -> { InfluxDbCpu clonedCpu = cpu.cloneAt(cpu.getTime().plusSeconds(60000l)); return InfluxDbWriteMessage.create(clonedCpu, NotUsed.notUsed()); }) .groupedWithin(10, Duration.of(50l, ChronoUnit.MILLIS)) .runWith(InfluxDbSink.typed(InfluxDbCpu.class, influxDB), materializer);
With QueryResult
source
Use InfluxDbSource.create
and InfluxDbSink.create
to create source and sink.
- Scala
-
val query = new Query("SELECT * FROM cpu", DatabaseName); val f1 = InfluxDbSource(influxDB, query) .map(resultToPoints) .runWith(InfluxDbSink.create())
- Java
-
Query query = new Query("SELECT * FROM cpu", DATABASE_NAME); CompletionStage<Done> completionStage = InfluxDbSource.create(influxDB, query) .map(queryResult -> points(queryResult)) .mapConcat(i -> i) .groupedWithin(10, Duration.of(50l, ChronoUnit.MILLIS)) .runWith(InfluxDbSink.create(influxDB), materializer);
TODO
Writing to InfluxDB
You can also build flow stages. InfluxDbFlow
. InfluxDbFlow
. The API is similar to creating Sinks.
- Scala
-
val result = Source( List( List(validMessage) ) ).via(InfluxDbFlow.create()) .runWith(Sink.seq) .futureValue
- Java
-
CompletableFuture<List<List<InfluxDbWriteResult<Point, NotUsed>>>> completableFuture = Source.single(Collections.singletonList(influxDbWriteMessage)) .via(InfluxDbFlow.create(influxDB)) .runWith(Sink.seq(), materializer) .toCompletableFuture();
Passing data through InfluxDbFlow
When streaming documents from Kafka, you might want to commit to Kafka AFTER the document has been written to InfluxDB.
- Scala
-
// We're going to pretend we got messages from kafka. // After we've written them to InfluxDB, we want // to commit the offset to Kafka case class KafkaOffset(offset: Int) case class KafkaMessage(cpu: InfluxDbFlowCpu, offset: KafkaOffset) val messagesFromKafka = List( KafkaMessage(new InfluxDbFlowCpu(Instant.now().minusSeconds(1000), "local_1", "eu-west-2", 1.4d, true, 123L), KafkaOffset(0)), KafkaMessage(new InfluxDbFlowCpu(Instant.now().minusSeconds(2000), "local_2", "eu-west-1", 2.5d, false, 125L), KafkaOffset(1)), KafkaMessage(new InfluxDbFlowCpu(Instant.now().minusSeconds(3000), "local_3", "eu-west-4", 3.1d, false, 251L), KafkaOffset(2)) ) var committedOffsets = List[KafkaOffset]() def commitToKafka(offset: KafkaOffset): Unit = committedOffsets = committedOffsets :+ offset val f1 = Source(messagesFromKafka) .map { kafkaMessage: KafkaMessage => val cpu = kafkaMessage.cpu println("hostname: " + cpu.getHostname) InfluxDbWriteMessage(cpu).withPassThrough(kafkaMessage.offset) } .groupedWithin(10, 50.millis) .via( InfluxDbFlow.typedWithPassThrough(classOf[InfluxDbFlowCpu]) ) .map { messages: Seq[InfluxDbWriteResult[InfluxDbFlowCpu, KafkaOffset]] => messages.foreach { message => commitToKafka(message.writeMessage.passThrough) } } .runWith(Sink.ignore)
- Java
-
// We're going to pretend we got metrics from kafka. // After we've written them to InfluxDb, we want // to commit the offset to Kafka /** Just clean the previous data */ influxDB.query(new Query("DELETE FROM cpu")); List<Integer> committedOffsets = new ArrayList<>(); List<MessageFromKafka> messageFromKafka = Arrays.asList( new MessageFromKafka( new InfluxDbCpu( Instant.now().minusSeconds(1000), "local_1", "eu-west-2", 1.4d, true, 123L), new KafkaOffset(0)), new MessageFromKafka( new InfluxDbCpu( Instant.now().minusSeconds(2000), "local_2", "eu-west-1", 2.5d, false, 125L), new KafkaOffset(1)), new MessageFromKafka( new InfluxDbCpu( Instant.now().minusSeconds(3000), "local_3", "eu-west-4", 3.1d, false, 251L), new KafkaOffset(2))); Consumer<KafkaOffset> commitToKafka = kafkaOffset -> committedOffsets.add(kafkaOffset.getOffset()); Source.from(messageFromKafka) .map( kafkaMessage -> { return InfluxDbWriteMessage.create( kafkaMessage.influxDbCpu, kafkaMessage.kafkaOffset); }) .groupedWithin(10, Duration.ofMillis(10)) .via(InfluxDbFlow.typedWithPassThrough(InfluxDbCpu.class, influxDB)) .map( messages -> { messages.stream() .forEach( message -> { KafkaOffset kafkaOffset = message.writeMessage().passThrough(); commitToKafka.accept(kafkaOffset); }); return NotUsed.getInstance(); }) .runWith(Sink.seq(), materializer) .toCompletableFuture() .get(10, TimeUnit.SECONDS);