Run Spark over an HBase Table

When we are running a MapReduce job on top of an HBase table, each mapper process a single region. You can also run a simple Java application that will scan the entire HBase table. This will be very similar with Spark. The HBase Spark API can return you an RDD representing your entire HBase table. But you partition this RDD to get it processed over multiple executors where each of them will process a single region

Calling HBase from Spark

The HBase Java API is accessible from the Spark code. Therefore, you can perform lookups like what we did when implementing our Flume interceptor. You can also use Spark to enrich and store data into HBase leveraging Puts and Increments. Last, the same way you can use MapReduce to generate HFiles for BulkLoads, you can use Spark to generate the same kind of files that you will later BulkLoad into HBase.

Spark and HBase: Puts

There are two ways to enrich HBase data while processing an input file. The first way is to do that in “real time,” where we update data into HBase as we are parsing the input file. This allows you to see the results in HBase as soon as they are read by Spark. You can also implement a stream processing pipeline using similar code and Spark streaming. Now, if you don’t have the requirement to get the data in real time in HBase, you can process it to generate HBase HFiles to be bulk loaded. While the streaming approach will give you a more real-time way of updating HBase, the bulk load approach provides better throughput, as it doesn’t need to interact with HBase.

The very first thing we will need is a text file big enough to get a sense of Spark’s performance capabilities. As in “Kafka”, we will create a small utility to generate this file. This utility can be called as follows:

java -classpath ~/ahae/target/ahae.jar:`hbase classpath`
  com.architecting.ch11.DataGenerator 100000 > ~/ahae/resources/ch11/data.txt

After the file has been generated, it must be pushed into HDFS for processing:

hdfs dfs -put  ~/ahae/resources/ch11/data.txt

Running Spark over a text file doesn’t require any specific HBase API. The code in Example 11-1 simply reads a file from HDFS and processes all the lines. To keep it simple, we will only count those lines here, but later on we will enrich this example.

 SparkConf sparkConf = new SparkConf().setAppName("JavaHBaseBulkGetExample ")
                                                             .setMaster("local[2]");
 JavaSparkContext jsc = new JavaSparkContext(sparkConf);
 JavaRDD<String> textFile = jsc.textFile("hdfs://localhost/user/cloudera/data.txt");
 JavaPairRDD<String, Integer> pairs = textFile.
                             mapToPair(new PairFunction<String, String, Integer>() {
   public Tuple2<String, Integer> call(String s) {
     return new Tuple2<String, Integer>(s.substring(0, s.indexOf("|")), 1); }
   });
 JavaPairRDD<String, Integer> counts = pairs.
                            reduceByKey(new Function2<Integer, Integer, Integer>() {
   public Integer call(Integer a, Integer b) { return a + b; }
   });
 System.out.println ("We have generaged " + counts.count() + " users");

This example should be straightforward—from each line, we extract the user ID only, then we group them together for a count. Because this is a very basic example, we will simply run it from Eclipse using the local environment.

Despite all the long DEBUG and INFO logs, what you are looking for is something like this:

We have generaged 999952 users

Now that we have a skeleton for our application, everything else will be standard HBase Java API calls mixed with Spark. As you might have figured, each entry in the input file represents information for someone. What we want to do is to update HBase with this information. There are two different ways to do that. The first option, illustrated in Example 11-2, has the benefit of being very easy to read and to understand. It will create one mutation per row on the input file and will buffer that to be sent to the HBase table.

SparkConf sparkConf = new SparkConf().setAppName("IngestLines ")
                                     .setMaster("local[2]");
JavaSparkContext jsc = new JavaSparkContext(sparkConf);
Configuration conf = HBaseConfiguration.create();
JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);
JavaRDD<String> textFile =
                   jsc.textFile("hdfs://localhost/user/cloudera/data.txt");

hbaseContext.bulkPut(textFile, TABLE_NAME, new Function<String, Put>() {
  @Override
  public Put call(String v1) throws Exception {
    String[] tokens = v1.split("\|");
    Put put = new Put(Bytes.toBytes(tokens[0]));
    put.addColumn(Bytes.toBytes("segment"),
                  Bytes.toBytes(tokens[1]),
                  Bytes.toBytes(tokens[2]));
    return put;
  }
});
jsc.close();

This code creates one mutation (a Put) for each and every column and simply returns it to the HBase Spark BulkPut framework. If we ignore the duplicates, we only have a maximum of seven columns per row, so it is acceptable to have one mutation for each. However, if you have tens of columns, you might improve performance by regrouping them first, and then creating the related Puts and emitting them directly to HBase. This approach is illustrated in Example 11-3, and if you are new to Spark, the code will be much more difficult to read and to understand. However, running on a local VM environment, with seven columns and one million lines, the code in Example 11-3 is about 10% faster than the code in Example 11-2. If you increase the number of columns, the difference will be even more significant.

SparkConf sparkConf = new SparkConf().setAppName("IngestLines ")
                                      .setMaster("local[2]");
JavaSparkContext jsc = new JavaSparkContext(sparkConf);
Configuration conf = HBaseConfiguration.create();
JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);
JavaRDD<String> textFile =
                   jsc.textFile("hdfs://localhost/user/cloudera/data.txt");

PairFunction<String, String, String> linesplit = 
new PairFunction<String, String, String>() {
  public Tuple2<String, String> call(String s) {
    int index = s.indexOf("|");
    return new Tuple2<String, String>(s.substring(0, index),
                                      s.substring(index + 1));
  }
};

JavaPairRDD<String, String> pairs = textFile.mapToPair(linesplit);
Function<String, List<String>> createCombiner =
new Function<String, List<String>>() {
  public List<String> call(String s) {
    List<String> list = new ArrayList<String>();
    list.add(s);
    return list;
  }
};

Function2<List<String>, String, List<String>> mergeValue =
new Function2<List<String>, String, List<String>>() {
  @Override
  public List<String> call(List<String> v1, String v2) throws Exception {
    v1.add(v2);
    return v1;
  }
};

Function2<List<String>, List<String>, List<String>> mergeCombiners =
new Function2<List<String>, List<String>, List<String>>() {
  @Override
  public List<String> call(List<String> v1, List<String> v2) throws Exception {
    v2.addAll(v1);
    return v2;
  }
};

JavaPairRDD<String, List<String>> keyValues = 
            pairs.combineByKey(createCombiner, mergeValue, mergeCombiners);

JavaRDD<Put> keyValuesPuts = keyValues.map( 
new Function<Tuple2<String, List<String>>, Put>() {
  @Override
  public Put call(Tuple2<String, List<String>> v1) throws Exception {
    Put put = new Put(Bytes.toBytes(v1._1));
    ListIterator<String> iterator = v1._2.listIterator();
    while (iterator.hasNext()) {
      String colAndVal = iterator.next();
      int indexDelimiter = colAndVal.indexOf("|");
      String columnQualifier = colAndVal.substring(0, indexDelimiter);
      String value = colAndVal.substring(indexDelimiter + 1);
      put.addColumn(COLUMN_FAMILY, Bytes.toBytes(columnQualifier),
                                   Bytes.toBytes(value));
    }
    return put;
  }
});

hbaseContext.foreachPartition(keyValuesPuts, 
  new VoidFunction<Tuple2<Iterator<Put>, Connection>>() {
    @Override
    public void call(Tuple2<Iterator<Put>, Connection> t) throws Exception {
      Table table = t._2().getTable(TABLE_NAME);
      BufferedMutator mutator = t._2().getBufferedMutator(TABLE_NAME);
      while (t._1().hasNext()) {
        Put put = t._1().next();
        mutator.mutate(put);
      }

      mutator.flush();
      mutator.close();
      table.close();
    }
  });
jsc.close();

Extract the key from the lines to create pairs that can be regrouped together.

Combine all the entries based on the key so each key will have a list of “column qualifier|value” strings associated.

Transform one key and all its related “column qualifier|value” strings to a single HBase put object.

Emit all the puts to HBase.

Basically, what we are doing here is regrouping all the lines based on the key that we extract from the string, then we transform them into a single put for this line that we then send to HBase.

You can run this example directly from Eclipse.

Because the output of this example is quite verbose, we will not reproduce it here. However, when it is done, you can query your HBase table to make sure data has been processed:

hbase(main):005:0> scan 'user', LIMIT => 2
ROW               COLUMN+CELL
 0000003542a7-... column=segment:postalcode, ts=1457057812022, value=34270
 0000013542a7-... column=segment:birthdate, ts=1457057756713, value=20/05/1946
2 row(s) in 0.0330 seconds

Similarly to what we have done in Chapter 9 to enrich data as we ingest it into HBase, you can use the default HBase API to perform lookups before creating the mutations and enrich them with the required information.

Spark on HBase: Bulk Load

In the previous section, we discussed a real-time approach of updating an HBase table. If this is not required for your use case, you will achieve a better throughput by using the HBase bulk load option. Let’s reuse the same example, but this time, instead of interacting directly with HBase, we will generate HBase HFiles that we will upload later on.

The data will remain the same. The main difference will be on the Spark side. HBase split the tables in regions based on the keys boundaries. Each HFile that we will create will have to belong to one of those regions and will have to contain keys only within those regions boundaries.

At the time of writing, the Java API to perform bulk loads in Spark is not completed. This is tracked by the JIRA HBASE-14217. Until this JIRA is resolved, only Scala can be used to perform this operation. Therefore, the code in Example 11-4 will be done in Scala.

import org.apache.hadoop.fs.Path
import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles
import org.apache.spark.SparkContext
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.spark._
import org.apache.hadoop.hbase.spark.HBaseContext
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.SparkConf
import org.apache.hadoop.hbase.spark.HBaseRDDFunctions._

object SparkBulkLoad {
  def main(args: Array[String]) {

    val columnFamily1 = "segment"
    val stagingFolder = "/tmp/user"
    val tableNameString = "user"
    val tableName = TableName.valueOf(tableNameString)

    val sc = new SparkContext(new SparkConf().setAppName("Spark BulkLoad")
                                             .setMaster("local[2]"))
    val config = HBaseConfiguration.create
    val hbaseContext = new HBaseContext(sc, config)

    val textFile = sc.textFile("hdfs://localhost/user/cloudera/data.txt")
    val toByteArrays = textFile.map(line => { 
      val tokens = line.split("\|")
      (Bytes.toBytes(tokens(0)), (Bytes.toBytes(columnFamily1),
                                  Bytes.toBytes(tokens(1)),
                                  Bytes.toBytes(tokens(2))))
    })

    toByteArrays.hbaseBulkLoad(hbaseContext, tableName, 
      t => { 
        val rowKey = t._1
        val family:Array[Byte] = t._2._1
        val qualifier = t._2._2
        val value = t._2._3

        val keyFamilyQualifier= new KeyFamilyQualifier(rowKey, family, qualifier)

        Seq((keyFamilyQualifier, value)).iterator
      },
      stagingFolder)

    val load = new LoadIncrementalHFiles(config)
    load.run(Array(stagingFolder, tableNameString)) 
  }
}

Apply transformation on the input file to transform each line into byte arrays representing the key, the column family, the column qualifier, and the value.

Based on the transformed input, make us of the Spark HBase API to generate the related HFiles.

Describe how bytes arrays are going to be transformed into the format required by the bulk load.

After HFiles are generated, make use of the LoadIncrementalHFiles API to load generated files into the target table.

There are a few steps required to run this application. First, because you will most probably not run this application as the Spark user, you will need permission to write in the applicationhistory directory. Use the following command to help you with this (you might need to set a password for HDFS):

sudo -u hdfs hadoop fs -chmod 1777 /user/spark/applicationHistory

Another important consideration to keep in mind is that, on a nonsecured cluster, HBase LoadIncrementalHFiles has to run with the HBase user. Indeed, it will push files into the /hbase tree, which will be read and written by the HBase processes. Pushing a file there with another user will end up with HBase not able to delete this file after compaction and fail.

You can run this small example using the following command:

su -l hbase -- spark-submit --class com.architecting.ch11.SparkBulkLoad 
            --master local ~/ahae/target/ahae.jar

Then add the following content into your HBase table:

hbase(main):013:0> scan 'user', LIMIT =>
ROW               COLUMN+CELL
 0000123542a7-... column=segment:postalcode, ts=1457567998199, value=34270
 0000153542a7-... column=segment:lastname, ts=1457567998199, value=Smith
 0000173542a7-... column=segment:birthdate, ts=1457567998199, value=06/03/1942
 0000173542a7-... column=segment:status, ts=1457567998199, value=maried

Spark Over HBase

The final example we will look at for using Spark and HBase will process the content of an HBase table using Spark. As we have done MapReduce over an HBase table, we will do Spark over an HBase table. The concept for Spark is similar; it will process the different regions in parallel.

The examples we’ll look at in this section are very simple, and they achieve the same goal—that is, they count the lines of an HBase table. Although this is very simple, you will see later how it can be extended for more complex use cases.

We’ll break the example down into a few pieces. The most important part is shown in Example 11-5.

  // SparkConf sc = new SparkConf().setAppName("ProcessTable").setMaster("local[2]");
  SparkConf sc = new SparkConf().setAppName("ProcessTable");
  JavaSparkContext jsc = new JavaSparkContext(sc);
  Configuration conf = HBaseConfiguration.create();

  JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);

  Scan scan = new Scan();
  scan.setCaching(100);
  KeyOnlyFilter kof = new KeyOnlyFilter();
  scan.setFilter(kof);

  JavaRDD<Tuple2<ImmutableBytesWritable, Result>> data =
                          hbaseContext.hbaseRDD(TABLE_NAME, scan);

This is where all the magic happens. We start with the normal Spark and HBase initialization calls. If you are used to HBase, and as we have seen when we have done MapReduce over an HBase table, we have to initialize a scan that will help us to filter the rows that we do not want and return the one we are looking at. Here we just want to get all the rows. However, because we want to count them, we do not need to get the value so we will filter it and keep only the keys. This should already be familiar to you. The last line is the interesting part. Using the HBaseContext and given the table name and the scan object, you will get an RDD representation of your HBase table. Processing this RDD will handle the entire HBase table. Example 11-6 is a very basic count operation on the table rows.

System.out.println("data.count() = " + data.count());

This simple line of code will trigger a count on the Spark RDD. The Spark RRD representing the entire HBase table, it will simply count all the rows in the table. This is very simple, but it doesn’t really let you do any specific operation on the rows.

Examples 11-7 and 11-8 also perform a count on the HBase table, however, as you will see, even if they just count the rows, it will be very easy to modify these examples to perform any manipulation of the row you might want.

  FlatMapFunction<Iterator<Tuple2<ImmutableBytesWritable, Result>>, Integer> setup =
  new FlatMapFunction<Iterator<Tuple2<ImmutableBytesWritable, Result>>, Integer>() {
    @Override
    public Iterable<Integer>
                      call(Iterator<Tuple2<ImmutableBytesWritable, Result>> input) {
      int a = 0;
      while (input.hasNext()) {
        a++; 
        input.next();
      }
      ArrayList<Integer> ret = new ArrayList<Integer>();
      ret.add(a);
      return ret;
    }
  };
  Function2<Integer, Integer, Integer> combine =
  new Function2<Integer, Integer, Integer>() {
    @Override
    public Integer call(Integer a, Integer b) {
      return a+b; 
    }
  };

  System.err.println("data.mapPartitions(setup).reduce(combine) = " +
      data.mapPartitions(setup).reduce(combine));
  long time3 = System.currentTimeMillis();
  System.err.println("Took " + (time3 - time2) + " milliseconds");

For each Spark partition, we simply increment a counter for each line we found. Here, it will be very easy to accumulate a value extracted from the HBase cell or do any other kind of aggregation we want.

Now that all the partitions are aggregated, to combine them, we simply need to sum all the values.

Example 11-7 looks very close to what we were doing in previous chapters with MapReduce. Because it doesn’t really map by partition, Example 11-8 is a bit different, but the output will again be the same.

  Function2<Integer, Tuple2<ImmutableBytesWritable, Result>, Integer> aggregator =
  new Function2<Integer, Tuple2<ImmutableBytesWritable, Result>, Integer>() {
    @Override
    public Integer call(Integer v1, Tuple2<ImmutableBytesWritable, Result> v2)
        throws Exception {
      return v1 + 1; 
    }
  };
  Function2<Integer, Integer, Integer> combiner =
  new Function2<Integer, Integer, Integer>() {
    @Override
    public Integer call(Integer v1, Integer v2) throws Exception {
      return v1 + v2; 
    }
  };

  System.err.println("data.aggregate(0, aggregator, combiner) = " +
                                           data.aggregate(0, aggregator, combiner));
  long time4 = System.currentTimeMillis();
  System.err.println("Took " + (time4 - time3) + " milliseconds");
  // end::PROCESS1[]

  jsc.close();
}

public static void main(String[] args) {
  processVersion1();
}

}

Because we just want to count the rows, we simply emit 1 for each one. However, think about something more complicated. Imagine that the cell contains an Avro object representing an order. If you want to sum all the order’s amounts, this is the place where you will do that. Simply extract the Avro object from the parameter and aggregate its amount.

Again, to combine the values together, we simply have to sum them.

Examples 11-7 and 11-8 can very easily be modified to achieve way more processing on the data you have:

• Think about doing data correlation where you perform an HBase lookup on the map side to enrich an output file that will generate Parquet data for analytic quests.

• Think about doing aggregation to enrich your HBase table with daily, weekly, and monthly values.

• Think about feeding a Kafka key to allow subsequent systems to receive and process the HBase data.

Everything you were previously doing with MapReduce can now be achieved using Spark.