MapR [SPARK-21] Structured Streaming MapR-DB Sink created (apache#219)

Author: rsotn-mapr

external/maprdb/src/main/scala/com/mapr/db/spark/streaming/DefaultSource.scala

@@ -0,0 +1,21 @@
+package com.mapr.db.spark.streaming
+
+import com.mapr.db.spark.streaming.sink.MapRDBSink
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.sql.execution.streaming.Sink
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.streaming.OutputMode
+
+class DefaultSource extends StreamSinkProvider with DataSourceRegister with Logging {
+
+  override def shortName(): String = "maprdb"
+
+  override def createSink(sqlContext: SQLContext,
+                          parameters: Map[String, String],
+                          partitionColumns: Seq[String],
+                          outputMode: OutputMode): Sink = {
+    new MapRDBSink(parameters)
+  }
+
+}

external/maprdb/src/main/scala/com/mapr/db/spark/streaming/MapRDBSourceConfig.scala

@@ -0,0 +1,12 @@
+package com.mapr.db.spark.streaming
+
+object MapRDBSourceConfig {
+
+  val Format: String = classOf[DefaultSource].getPackage.getName
+  val TablePathOption: String = "tablePath"
+  val IdFieldPathOption: String = "idFieldPath"
+  val CreateTableOption: String = "createTable"
+  val BulkModeOption: String = "bulkMode"
+  val SampleSizeOption: String = "sampleSize"
+
+}

external/maprdb/src/main/scala/com/mapr/db/spark/streaming/sink/MapRDBSink.scala

@@ -0,0 +1,34 @@
+package com.mapr.db.spark.streaming.sink
+
+
+import com.mapr.db.spark.sql._
+import com.mapr.db.spark.streaming.MapRDBSourceConfig
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.execution.streaming.Sink
+import org.ojai.DocumentConstants
+
+private[streaming] class MapRDBSink(parameters: Map[String, String]) extends Sink with Logging {
+
+  @volatile private var latestBatchId = -1L
+
+  override def toString(): String = "MapRDBSink"
+
+  override def addBatch(batchId: Long, data: DataFrame): Unit = {
+    if (batchId <= latestBatchId) {
+      logInfo(s"Skipping already committed batch $batchId")
+    } else {
+
+      val tablePath = parameters.get(MapRDBSourceConfig.TablePathOption)
+      require(tablePath.isDefined, s"'${MapRDBSourceConfig.TablePathOption}' option must be defined")
+
+      val idFieldPath = parameters.getOrElse(MapRDBSourceConfig.IdFieldPathOption, DocumentConstants.ID_KEY)
+      val createTable = parameters.getOrElse(MapRDBSourceConfig.CreateTableOption, "false").toBoolean
+      val bulkInsert = parameters.getOrElse(MapRDBSourceConfig.BulkModeOption, "false").toBoolean
+
+      data.saveToMapRDB(tablePath.get, idFieldPath, createTable, bulkInsert)
+
+      latestBatchId = batchId
+    }
+  }
+}