chronicler-spark

Open-source InfluxDB connector for Apache Spark on top of Chronicler.

Get Started

At the beginning add required module to your build.sbt:

// For RDD
libraryDependencies += "com.github.fsanaulla" %% "chronicler-spark-rdd" % <version>

// For Dataset
libraryDependencies += "com.github.fsanaulla" %% "chronicler-spark-ds" % <version>

// For Structured Streaming
libraryDependencies += "com.github.fsanaulla" %% "chronicler-spark-structured-streaming" % <version>

// For DStream
libraryDependencies += "com.github.fsanaulla" %% "chronicler-spark-streaming" % <version>

Usage

Default configuration:

final case class InfluxConfig(
    host: String,
    port: Int = 8086,
    credentials: Option[InfluxCredentials] = None,
    compress: Boolean = false,
    ssl: Boolean = false)

It's recommended to enable data compression to decrease network traffic.

For RDD[T]:

import com.github.fsanaulla.chronicler.spark.rdd._

val rdd: RDD[T] = _
rdd.saveToInfluxDBMeas("dbName", "measurementName")

// to save with dynamicly generated measurement
rdd.saveToInfluxDB("dbName")

For Dataset[T]:

import com.github.fsanaulla.chronicler.spark.ds._

val ds: Dataset[T] = _
ds.saveToInfluxDBMeas("dbName", "measurementName")

// to save with dynamicly generated measurement
ds.saveToInfluxDB("dbName")

For DataStreamWriter[T]

import com.github.fsanaulla.chronicler.spark.structured.streaming._

val structStream: DataStreamWriter[T] = _
val saved = structStream.saveToInfluxDBMeas("dbName", "measurementName")

// to save with dynamicly generated measurement
val saved = structStream.saveToInfluxDB("dbName")
..
saved.start().awaitTermination()

For DStream[T]:

import com.github.fsanaulla.chronicler.spark.streaming._

val stream: DStream[T] = _
stream.saveToInfluxDBMeas("dbName", "measurementName")
stream,saveToInfluxDB("dbName")