Writing Spark DataFrame to HBase Table using Hortonworks

This tutorial explains how to insert/write Spark DataFrame rows to HBase table using Hortonworks DataSource "org.apache.spark.sql.execution.datasources.hbase" from shc-core library with Scala example.

In order to connect Spark with HBase, you would need the following API's.

hbase-client library which natively interacts with HBase.
hbase-spark connector which provides HBaseContext to interact Spark with HBase. HBaseContext pushes the configuration to the Spark executors and allows it to have an HBase Connection per Spark Executor.
shc-core DataSource API to work with DataFrames on HBase table.

For better compatibility let's use all these from Hortonworks, at the time on writing this article, versions provided in below dependencies are latest.


  <dependencies>
    <dependency>
      <groupId>org.apache.hbase</groupId>
      <artifactId>hbase-client</artifactId>
      <version>2.0.2.3.1.0.6-1</version> <!-- Hortonworks Latest -->
    </dependency>
    <dependency>
      <groupId>org.apache.hbase</groupId>
      <artifactId>hbase-spark</artifactId>
      <version>2.0.2.3.1.0.6-1</version> <!-- Hortonworks Latest -->
    </dependency>
    <dependency>
      <groupId>com.hortonworks</groupId>
      <artifactId>shc-core</artifactId>
      <version>1.1.1-2.1-s_2.11</version> <!-- Hortonworks Latest -->
    </dependency>
  </dependencies>

Note that hbase-spark provides spark-core and spark-sql API's version 2.1.1 as transitive dependencies hence, we don't have to provide these additionally.

Below is the complete example for your reference and this is available at GitHub.


package com.aistudylabs.spark.dataframe.hbase.hortonworks

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.datasources.hbase.HBaseTableCatalog

object HBaseSparkExample {

  case class Employee(key: String, fName: String, lName: String,
                      mName:String, addressLine:String, city:String,
                      state:String, zipCode:String)

  def main(args: Array[String]): Unit = {

    def catalog =
      s"""{
         |"table":{"namespace":"default", "name":"employee"},
         |"rowkey":"key",
         |"columns":{
         |"key":{"cf":"rowkey", "col":"key", "type":"string"},
         |"fName":{"cf":"person", "col":"firstName", "type":"string"},
         |"lName":{"cf":"person", "col":"lastName", "type":"string"},
         |"mName":{"cf":"person", "col":"middleName", "type":"string"},
         |"addressLine":{"cf":"address", "col":"addressLine", "type":"string"},
         |"city":{"cf":"address", "col":"city", "type":"string"},
         |"state":{"cf":"address", "col":"state", "type":"string"},
         |"zipCode":{"cf":"address", "col":"zipCode", "type":"string"}
         |}
         |}""".stripMargin

    val data = Seq(Employee("1","Abby","Smith","K","3456 main", "Orlando","FL","45235"),
      Employee("2","Amaya","Williams","L","123 Orange","Newark","NJ", "27656"),
      Employee("3","Alchemy","Davis","P","Warners","Sanjose","CA", "34789"))

    val spark: SparkSession = SparkSession.builder()
      .master("local[1]")
      .appName("AIStudyLabs.com")
      .getOrCreate()

    import spark.implicits._
    val df = spark.sparkContext.parallelize(data).toDF

    df.write.options(
      Map(HBaseTableCatalog.tableCatalog -> catalog, HBaseTableCatalog.newTable -> "4"))
      .format("org.apache.spark.sql.execution.datasources.hbase")
      .save()
  }
}

Let me explain what's happening at a few statements in this example.

We define a catalog which bridges the gap between HBase KV store and Spark DataFrame table structure. using this we will also map the column names between the two structures and keys.


df.write.options(
      Map(HBaseTableCatalog.tableCatalog -> catalog, HBaseTableCatalog.newTable -> "4"))
      .format("org.apache.spark.sql.execution.datasources.hbase")
      .save()

A couple of things happening at this snippet. format takes "org.apache.spark.sql.execution.datasources.hbase" DataSource defined in "shc-core" API which enables us to use DataFrame with HBase tables. And, df.write.options take the catalog and specifies to use 4 regions in cluster. Finally, save()writes it to HBase table.

If you use scan 'employee' on a shell, you will get the below output

Next, we will see how to access HBase table from Spark DataFrame.

Conclusion:

In this tutorial, you have learned how to read from and write Spark DataFrame to HBase table using Hortonworks DataSource API. Also, you have learned all different libraries involved maven dependencies.

Writing Spark DataFrame to HBase Table using Hortonworks

Conclusion:

Related Articles