Introducing the ReactiveInflux: non-blocking InfluxDB driver for Scala and Java supporting Apache Spark

Rado BuranskyApril 10th, 2016Last Updated: April 8th, 2016

0 110 2 minutes read

I am excited to announce a very first release of ReactiveInflux developed at Pygmalios. InfluxDB missed a non-blocking driver for both Scala and Java. Immutability, testability and extensibility are key features of ReactiveInflux. Comming with a support for Apache Spark it is the weapon of choice.

https://github.com/pygmalios/reactiveinflux

It internally uses Play Framework WS API which is a rich asynchronous HTTP client built on top of Async Http Client.

Features

asynchronous (non-blocking) interface for Scala
synchronous (blocking) interface for Scala and Java
supports both Spark and Spark streaming
immutability
testability
extensibility

Compatibility

InfluxDB 0.11, 0.10 and 0.9 (maybe even older too)
Scala 2.11 and 2.10
Java 7 and above
Apache Spark 1.4 and above

Scala asynchronous (non-blocking) example

val result = withInfluxDb(new URI("http://localhost:8086/"), "example1") { db =>
  db.create().flatMap { _ =>
    val point = Point(
      time        = DateTime.now(),
      measurement = "measurement1",
      tags        = Map("t1" -> "A", "t2" -> "B"),
      fields      = Map(
        "f1" -> 10.3,
        "f2" -> "x",
        "f3" -> -1,
        "f4" -> true)
    )
    db.write(point).flatMap { _ =>
      db.query("SELECT * FROM measurement1").flatMap { queryResult =>
        println(queryResult.row.mkString)
        db.drop()
      }
    }
  }
}

Scala synchronous (blocking) example

implicit val awaitAtMost = 10.seconds
syncInfluxDb(new URI("http://localhost:8086/"), "example1") { db =>
  db.create()
 
  val point = Point(
    time        = DateTime.now(),
    measurement = "measurement1",
    tags        = Map("t1" -> "A", "t2" -> "B"),
    fields      = Map(
      "f1" -> 10.3,
      "f2" -> "x",
      "f3" -> -1,
      "f4" -> true)
  )
  db.write(point)
 
  val queryResult = db.query("SELECT * FROM measurement1")
  println(queryResult.row.mkString)
 
  db.drop()
}

Java synchronous (blocking) example

// Use Influx at the provided URL
ReactiveInfluxConfig config = new JavaReactiveInfluxConfig(
  new URI("http://localhost:8086/"));
long awaitAtMostMillis = 30000;
try (SyncReactiveInflux reactiveInflux = new JavaSyncReactiveInflux(
  config, awaitAtMostMillis)) {
    SyncReactiveInfluxDb db = reactiveInflux.database("example1");
    db.create();
 
    Map tags = new HashMap<>();
    tags.put("t1", "A");
    tags.put("t2", "B");
 
    Map fields = new HashMap<>();
    fields.put("f1", 10.3);
    fields.put("f2", "x");
    fields.put("f3", -1);
    fields.put("f4", true);
 
    Point point = new JavaPoint(
        DateTime.now(),
        "measurement1",
        tags,
        fields
    );
    db.write(point);
 
    QueryResult queryResult = db.query("SELECT * FROM measurement1");
    System.out.println(queryResult.getRow().mkString());
 
    db.drop();
}

Apache Spark Scala example

val point1 = Point(
  time        = DateTime.now(),
  measurement = "measurement1",
  tags        = Map(
    "tagKey1" -> "tagValue1",
    "tagKey2" -> "tagValue2"),
  fields      = Map(
    "fieldKey1" -> "fieldValue1",
    "fieldKey2" -> 10.7)
)
sc.parallelize(Seq(point1)).saveToInflux()

Apache Spark streaming Scala example

val point1 = Point(
  time        = DateTime.now(),
  measurement = "measurement1",
  tags        = Map(
    "tagKey1" -> "tagValue1",
    "tagKey2" -> "tagValue2"),
  fields      = Map(
    "fieldKey1" -> "fieldValue1",
    "fieldKey2" -> 10.7)
)
val queue = new mutable.Queue[RDD[Point]]
queue.enqueue(ssc.sparkContext.parallelize(Seq(point1)))
ssc.queueStream(queue).saveToInflux()

Apache Spark Java example

...
SparkInflux sparkInflux = new SparkInflux("example", 1000);
sparkInflux.saveToInflux(sc.parallelize(Collections.singletonList(point)));

Apache Spark streaming Java example

...
SparkInflux sparkInflux = new SparkInflux("example", 1000);
Queue> queue = new LinkedList<>();
queue.add(ssc.sparkContext().parallelize(Collections.singletonList(point)));
sparkInflux.saveToInflux(ssc.queueStream(queue));

Credit to Pygmalios

Top-tech startup based in Bratislava, Slovakia invests into cutting edge technologies to ensure rapid growth in the domain of real-time predictive retail analytics.

Reference:

Introducing the ReactiveInflux: non-blocking InfluxDB driver for Scala and Java supporting Apache Spark from our JCG partner Rado Buransky at the Rado Buransky’s Blog blog.

Do you want to know how to develop your skillset to become a Java Rockstar?

Subscribe to our newsletter to start Rocking right now!

To get you started we give you our best selling eBooks for FREE!

1. JPA Mini Book

2. JVM Troubleshooting Guide

3. JUnit Tutorial for Unit Testing

4. Java Annotations Tutorial

5. Java Interview Questions

6. Spring Interview Questions

7. Android UI Design