文档首页/ MapReduce服务 MRS/ 开发指南（普通版_3.x）/ Spark2x开发指南（普通模式）/ 开发Spark应用/ Spark读取HBase表样例程序/ 分布式Scan HBase表
更新时间：2024-06-27 GMT+08:00
查看PDF
分享

分布式Scan HBase表

场景说明

用户可以在Spark应用程序中使用HBaseContext的方式去操作HBase，使用hbaseRDD方法以特定的规则扫描HBase表。

数据规划

使用操作Avro格式数据章节中创建的HBase数据表。

开发思路

  1. 设置scan的规则，例如：setCaching。
  2. 使用特定的规则扫描Hbase表。

打包项目

  • 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考在Linux环境中编包并运行Spark程序。
  • 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“$SPARK_HOME” ）下。

若运行“Spark on HBase”样例程序，需要在Spark客户端的“spark-defaults.conf”配置文件中将配置项“spark.yarn.security.credentials.hbase.enabled”设置为“true”（该参数值默认为“false”，改为“true”后对已有业务没有影响。如果要卸载HBase服务，卸载前请将此参数值改回“false”），将配置项“spark.inputFormat.cache.enabled”设置为“false”。

提交命令

假设用例代码打包后的jar包名为spark-hbaseContext-test-1.0.jar，并将jar包放在客户端$SPARK_HOME目录下，以下命令均在$SPARK_HOME目录执行，Java接口对应的类名前有Java字样，请参考具体样例代码进行书写。

  • yarn-client模式：

    java/scala版本（类名等请与实际代码保持一致，此处仅为示例）

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

    python版本（文件名等请与实际保持一致，此处仅为示例）

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

  • yarn-cluster模式：

    java/scala版本（类名等请与实际代码保持一致，此处仅为示例）

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

    python版本（文件名等请与实际保持一致，此处仅为示例）

    bin/spark-submit --master yarn --deploy-mode cluster --jars SparkOnHbaseJavaExample-1.0.jar HBaseDistributedScanExample.py ExampleAvrotable

Java样例代码

下面代码片段仅为演示，具体代码参见SparkOnHbaseJavaExample中JavaHBaseDistributedScanExample文件：

 public static void main(String[] args) throws IOException{
    if (args.length < 1) {
      System.out.println("JavaHBaseDistributedScan {tableName}");
      return;
    }
    String tableName = args[0];
    SparkConf sparkConf = new SparkConf().setAppName("JavaHBaseDistributedScan " + tableName);
    JavaSparkContext jsc = new JavaSparkContext(sparkConf);
    try {
      Configuration conf = HBaseConfiguration.create();
      JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);
      Scan scan = new Scan();
      scan.setCaching(100);
      JavaRDD<Tuple2<ImmutableBytesWritable, Result>> javaRdd =
              hbaseContext.hbaseRDD(TableName.valueOf(tableName), scan);
      List<String> results = javaRdd.map(new ScanConvertFunction()).collect();
      System.out.println("Result Size: " + results.size());
    } finally {
      jsc.stop();
    }
  }

Scala样例代码

下面代码片段仅为演示，具体代码参见SparkOnHbaseScalaExample中HBaseDistributedScanExample文件：

  def main(args: Array[String]) {
    if (args.length < 1) {
      println("HBaseDistributedScanExample {tableName} missing an argument")
      return
    }
    val tableName = args(0)
    val sparkConf = new SparkConf().setAppName("HBaseDistributedScanExample " + tableName )
    val sc = new SparkContext(sparkConf)
    try {
      val conf = HBaseConfiguration.create()
      val hbaseContext = new HBaseContext(sc, conf)
      val scan = new Scan()
      scan.setCaching(100)
      val getRdd = hbaseContext.hbaseRDD(TableName.valueOf(tableName), scan)
      getRdd.foreach(v => println(Bytes.toString(v._1.get())))
      println("Length: " + getRdd.map(r => r._1.copyBytes()).collect().length);
    } finally {
      sc.stop()
    }
  }

Python样例代码

下面代码片段仅为演示，具体代码参见SparkOnHbasePythonExample中HBaseDistributedScanExample文件：

# -*- coding:utf-8 -*-
# -*- coding:utf-8 -*-
"""
【说明】
由于pyspark不提供Hbase相关api,本样例使用Python调用Java的方式实现
"""
from py4j.java_gateway import java_import
from pyspark.sql import SparkSession
# 创建SparkSession
spark = SparkSession\
        .builder\
        .appName("JavaHBaseDistributedScan")\
        .getOrCreate()
# 向sc._jvm中导入要运行的类
java_import(spark._jvm, 'com.huawei.bigdata.spark.examples.hbasecontext.JavaHBaseDistributedScanExample')
# 创建类实例并调用方法，传递sc._jsc参数
spark._jvm.JavaHBaseDistributedScan().execute(spark._jsc, sys.argv)
# 停止SparkSession
spark.stop()

相关文档