Using the BulkGet Interface

Scenario

Users can use the HBaseContext method to use HBase in Spark applications, construct the rowkey of the data to be obtained into RDDs, and obtain the data corresponding to the rowkey in the HBase tables through the BulkGet interface of HBaseContext.

Data Planning

Perform operations based on the HBase tables and data in the tables that are created in 3.5.3 Using the BulkPut Interface.

Development Guideline

1. Create RDDs containing the rowkey to be obtained.
2. Perform operations on HBase in HBaseContext mode and obtain data corresponding to rowkey in HBase tables through the BulkGet interface of HBaseContext.

Packaging the Project

• Use the Maven tool provided by IDEA to pack the project and generate a JAR file. For details, see Compiling and Running the Application.
• Upload the JAR package to any directory (for example, $SPARK_HOME) on the server where the Spark client is located.

To run the Spark on HBase example program, set spark.yarn.security.credentials.hbase.enabled (false by default) in the spark-defaults.conf file on the Spark client to true. Changing the spark.yarn.security.credentials.hbase.enabled value does not affect existing services. (To uninstall the HBase service, you need to change the value of this parameter back to false.) Set the value of the configuration item spark.inputFormat.cache.enabled to false.

Submitting Commands

Assume that the JAR package name is spark-hbaseContext-test-1.0.jar that is stored in the $SPARK_HOME directory on the client. The following commands are executed in the $SPARK_HOME directory, and Java is displayed before the class name of the Java interface. For details, see the sample code.

• yarn-client mode:

  Java/Scala version (The class name must be the same as the actual code. The following is only an example.)

  bin/spark-submit --master yarn --deploy-mode client --class com.huawei.bigdata.spark.examples.hbasecontext.JavaHBaseBulkGetExample SparkOnHbaseJavaExample-1.0.jar bulktable

  Python version. (The file name must be the same as the actual one. The following is only an example.)

  bin/spark-submit --master yarn --deploy-mode client --jars SparkOnHbaseJavaExample-1.0.jar HBaseBulkGetExample.py bulktable

• yarn-cluster mode:

  Java/Scala version (The class name must be the same as the actual code. The following is only an example.)

  bin/spark-submit --master yarn --deploy-mode cluster --class com.huawei.bigdata.spark.examples.hbasecontext.JavaHBaseBulkGetExample SparkOnHbaseJavaExample-1.0.jar bulktable

  Python version. (The file name must be the same as the actual one. The following is only an example.)

  bin/spark-submit --master yarn --deploy-mode cluster --jars spark-hbase-python-test-1.0.jar HBaseBulkGetExample.py bulktable

Java Sample Code

The following code snippet is only for demonstration. For details about the code, see the HBaseBulkGetExample file in SparkOnHbaseJavaExample.

 public static void main(String[] args) throws IOException{
    if (args.length < 1) {
      System.out.println("JavaHBaseBulkGetExample  {tableName}");
      return;
    }
    String tableName = args[0];
    SparkConf sparkConf = new SparkConf().setAppName("JavaHBaseBulkGetExample " + tableName);
    JavaSparkContext jsc = new JavaSparkContext(sparkConf);
    try {
      List<byte[]> list = new ArrayList<byte[]>(5);
      list.add(Bytes.toBytes("1"));
      list.add(Bytes.toBytes("2"));
      list.add(Bytes.toBytes("3"));
      list.add(Bytes.toBytes("4"));
      list.add(Bytes.toBytes("5"));
      JavaRDD<byte[]> rdd = jsc.parallelize(list);
      Configuration conf = HBaseConfiguration.create();
      JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);
      List resultList = hbaseContext.bulkGet(TableName.valueOf(tableName), 2, rdd, new GetFunction(),
              new ResultFunction()).collect();
      for(int i =0 ;i<resultList.size();i++){
        System.out.println(resultList.get(i));
      }
    } finally {
      jsc.stop();
    }
  }

Scala Sample Code

The following code snippet is only for demonstration. For details about the code, see the HBaseBulkGetExample file in SparkOnHbaseScalaExample.

 def main(args: Array[String]) {
    if (args.length < 1) {
      println("HBaseBulkGetExample {tableName} missing an argument")
      return
    }
    val tableName = args(0)
    val sparkConf = new SparkConf().setAppName("HBaseBulkGetExample " + tableName)
    val sc = new SparkContext(sparkConf)
    try {
      //[(Array[Byte])]
      val rdd = sc.parallelize(Array(
        Bytes.toBytes("1"),
        Bytes.toBytes("2"),
        Bytes.toBytes("3"),
        Bytes.toBytes("4"),
        Bytes.toBytes("5"),
        Bytes.toBytes("6"),
        Bytes.toBytes("7")))
      val conf = HBaseConfiguration.create()
      val hbaseContext = new HBaseContext(sc, conf)
      val getRdd = hbaseContext.bulkGet[Array[Byte], String](
        TableName.valueOf(tableName),
        2,
        rdd,
        record => {
          System.out.println("making Get")
          new Get(record)
        },
        (result: Result) => {
          val it = result.listCells().iterator()
          val b = new StringBuilder
          b.append(Bytes.toString(result.getRow) + ":")
          while (it.hasNext) {
            val cell = it.next()
            val q = Bytes.toString(CellUtil.cloneQualifier(cell))
            if (q.equals("counter")) {
              b.append("(" + q + "," + Bytes.toLong(CellUtil.cloneValue(cell)) + ")")
            } else {
              b.append("(" + q + "," + Bytes.toString(CellUtil.cloneValue(cell)) + ")")
            }
          }
          b.toString()
        })
      getRdd.collect().foreach(v => println(v))
    } finally {
      sc.stop()
    }
  }

Python Sample Code

The following code snippet is only for demonstration. For details about the code, see the HBaseBulkGetExample file in SparkOnHbasePythonExample.

# -*- coding:utf-8 -*- 
 """ 
 [Note] 
 PySpark does not provide HBase-related APIs. In this example, Python is used to invoke Java code to implement required operations.
  """ 
 from py4j.java_gateway import java_import 
 from pyspark.sql import SparkSession 
 # Create a SparkSession instance. 
 spark = SparkSession\ 
         .builder\ 
         .appName("JavaHBaseBulkGetExample")\ 
         .getOrCreate() 
 # Import required class to sc._jvm.
 java_import(spark._jvm, 'com.huawei.bigdata.spark.examples.hbasecontext.JavaHBaseBulkGetExample') 
 # Create a class instance and invoke the method. Transfer the sc._jsc parameter. 
 spark._jvm.JavaHBaseBulkGetExample().execute(spark._jsc, sys.argv) 
 # Stop the SparkSession instance. 
 spark.stop()