Transaction Support and Smart Joins for MapR Database

Contributed by

10 min read

Sometimes there are limits around how much we can stretch certain technology. In the case of MapR Database, these limits never seem to get closer, while we add more and more capabilities on top it.

Previously, we have talked about many things we can do using MapR Database. Make sure you check these related posts.

Today, we want to introduce the idea of transactional writing when using MapR Database.

This is not something supported out of the box by this distributed database; however, when using Apache Spark, we could implement similar concepts to what relational databases have.

It is not until recently that Spark added APIs to start supporting these ideas, so today we are going to review some of these APIs while proposing a way to add transactions to MapR Database.

Certainly, the transactional context is, in our case, at the application layer, so there are only a few things we can actually do. Apache Spark proposes it as best effort since this is a very fragile context and many, many things can go wrong.

Let's review what Apache Spark API offers in order to support transactional writes.

The most basic building block is called DataWriter[Row], and it is defined as follows:

class MapRDBDataWriter extends DataWriter[Row] with Logging {
  override def write(record: Row): Unit = ???

  override def commit(): WriterCommitMessage = ???

  override def abort(): Unit = ???

A DataWriter[Row] is in charge of writing a particular partition of distributed data in Spark to the target source.

The write function receives individual records to be written down to our target source, in our case, MapR Database.

The commit function is called once all records of a partition have been successfully written down.

abort is called if a record fails to be written down.

Putting it in context, in order for a transaction to happen, all partitions must successfully commit, but at the partition level, it is impossible to know what has happened to other partitions. In other words, the transaction must be processed in two phases: in the first phase, a partition commits correctly (task level), and in the second phase, all partitions are successfully committed at the job level. If anything fails, the transaction fails, and we must provide a way to roll it back.


The MapRDBDataWriterFactory is in charge of creating multiple DataWriters. This class looks like this:

class MapRDBDataWriterFactory(table: String, schema: StructType) extends DataWriterFactory[Row] {
    override def createDataWriter(partitionId: Int, attemptNumber: Int): DataWriter[Row] = new MapRDBDataWriter(...)

A DataWriter does the heavy work of writing particular records, belonging to a partition, down to MapR Database.

It is important to notice that Spark might call createDataWriter many times for the same partitionId. This happens if a particular task is slow or the task fails. Spark creates a new DataWriter with a different attemptNumber, which implies that many writers might try to write the same data down.

Our implementation looks like this:

class MapRDBDataWriterFactory(table: String, schema: StructType) extends DataWriterFactory[Row] {

  @transient private lazy val connection = DriverManager.getConnection("ojai:mapr:")

  @transient private lazy val store: DocumentStore = connection.getStore(table)

  private val writtenIds = scala.collection.mutable.ListBuffer.empty[String]

  override def createDataWriter(partitionId: Int, attemptNumber: Int): DataWriter[Row] = new DataWriter[Row] with Logging {"PROCESSING PARTITION ID: $partitionId ; ATTEMPT: $attemptNumber")

    override def write(record: Row): Unit = {

      val doc = schema
        .map(field => (, schema.fieldIndex(
        .foldLeft(connection.newDocumentBuilder()) { case (acc, (name, idx)) => acc.put(name, record.getString(idx)) }

      this.synchronized {
        if (!writtenIds.contains(doc.getIdString)) {

    override def commit(): WriterCommitMessage = {"PARTITION $partitionId COMMITTED AFTER ATTEMPT $attemptNumber")

      CommittedIds(partitionId, writtenIds.toSet)

    override def abort(): Unit = {"PARTITION $partitionId ABORTED AFTER ATTEMPT $attemptNumber")

      MapRDBCleaner.clean(writtenIds.toSet, table)"PARTITION $partitionId CLEANED UP")

Notice that in the write function, given a Row and the corresponding Schema, we build an OJAI Document and insert it into MapR Database. Then we save the _ids, so we can roll back the data written in this partition if something goes wrong.

We have optimized it a little bit, so records are not written twice by having a shared state with the _ids of already written records.

The abort function does exactly what we just described. If it is called, it deletes the already written records (rollback).

commit informs back to the driver that all records for the partition have been written and the partition has been committed. If there are other tasks for the same partition, the driver will ignore the commit messages after the first one.


The MapRDBDataSourceWriter runs at the driver, and it is in charge of collecting and controlling the results of each partition.

class MapRDBDataSourceWriter(table: String, schema: StructType) extends DataSourceWriter with Logging {

    override def createWriterFactory(): DataWriterFactory[Row] = ???

    override def commit(messages: Array[WriterCommitMessage]): Unit = ???

    override def abort(messages: Array[WriterCommitMessage]): Unit = ???


The createWriterFactory creates the DataWriterFactory that runs on the executor side.

override def createWriterFactory(): DataWriterFactory[Row] = new MapRDBDataWriterFactory(table, schema)

commits gets the commit messages from each of the DataWriters. If all commits are successful, then the entire job is successful, and we are good to go; the transaction has finished.

If there is at least one partition that was not successfully committed, the job has failed. The failed partitions know how to roll back themselves (explained above), but the driver must roll back any other data from successful partitions. In order to do this, we collect all successfully committed _ids from all committed partitions, and we use them in the rollback phase.

override def commit(messages: Array[WriterCommitMessage]): Unit = {
    val ids = messages.foldLeft(Set.empty[String]) { case (acc, CommittedIds(partitionId, partitionIds)) =>"PARTITION $partitionId HAS BEEN CONFIRMED BY DRIVER")

      acc ++ partitionIds

    // Let's make sure this is thread-safe
    globallyCommittedIds = this.synchronized {
      globallyCommittedIds ++ ids

If a partition fails, then abort in the DataWriter is called (explained above), so partition data is rolled back. The abort in the driver is called, so we can roll back any other written data.

override def abort(messages: Array[WriterCommitMessage]): Unit = {"JOB BEING ABORTED")"JOB CLEANING UP")

    MapRDBCleaner.clean(globallyCommittedIds.toSet, table)


WriteSupport is the entry point for injecting our code into Spark.

class Writer extends WriteSupport with Logging {

  override def createWriter(jobId: String, schema: StructType, mode: SaveMode, options: DataSourceOptions): Optional[DataSourceWriter] = {

    val tablePath = options.get("path").get()"TABLE PATH BEING USED: $tablePath")

    java.util.Optional.of(new MapRDBDataSourceWriter(tablePath, schema))

Using Transaction Writes

val df: DataFrame = ...


Of course, we are wrapping this into a better API, so we can do the following:

data.writeToMapRDB("/user/mapr/tables/my_table", withTransaction = true)

By indicating withTransaction = true, Spark tries its best to write the given DataFrame in a transactional mode using the described mechanics above. If withTransaction = false, then we use the regular, official MapR Database Connector for Apache Spark to write the DataFrame down.

This feature has been introduced in Experimental mode, so we can try it out and decide if it is worth having.

Bonus: Smart Join

Along with transactional writing, we added a new experimental feature that we called smart join. The idea is that we could join any DataFrame with a MapR Database table that has not been loaded yet.

Let's quickly look at what this API looks like:

val rdd = sparkSession
    .parallelize(1 to 1000000)
    .map(n => Row(n.toString))

val df = sparkSession
    .createDataFrame(rdd, new StructType().add("value", StringType))

val joint = df
      new StructType().add("payload", StringType),


 |-- value: string (nullable = true)
 |-- payload: string (nullable = true)

As we can see, df is a DataFrame where we can call joinWithMapRDBTable. What really is happening here is that only those rows from the MapR Database table that actually join with df will ever be fetched. This is in contrast to the traditional process of loading the table and then doing the join.

In our case, we look at the join condition and filter, using secondary indexes if possible, the MapR Database table at the right side of the join. This considerably reduces the amount of data transferred to Spark. It's important to notice that we are also doing projections pushdown, so we can do additional optimizations to reduce data transferring.


This blog post was published April 03, 2019.

50,000+ of the smartest have already joined!

Stay ahead of the bleeding edge...get the best of Big Data in your inbox.

Get our latest posts in your inbox

Subscribe Now