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.