# SparkStreaming批量写入HBase表
#### 场景说明
用户可以在Spark应用程序中使用HBaseContext的方式去操作HBase，使用streamBulkPut接口将流数据写入HBase表中。
#### 数据规划
1. 在客户端执行**hbase shell**进入HBase命令行。
2. 在hbase命令执行下面的命令创建HBase表： **create** *'streamingTable','cf1'*
   
3. 在客户端另外一个session通过linux命令构造一个端口进行接收数据（不同操作系统的机器，命令可能不同，suse尝试使用netcat -lk 9999）： **nc -lk** *9999*
   提交任务命令执行之后，在该命令下输入要提交的数据，通过HBase表进行接收。
   ![](https://support.huaweicloud.com/devg-lts-mrs/public_sys-resources/note_3.0-zh-cn.png)
   在构造一个端口进行接收数据时，需要在客户端所在服务器上安装netcat。
   
 
#### 开发思路
1. 使用SparkStreaming持续读取特定端口的数据。
2. 将读取到的Dstream通过streamBulkPut接口写入HBase表中。
 
#### 运行前置操作
安全模式下Spark Core样例代码需要读取两个文件（user.keytab、krb5.conf）。user.keytab和krb5.conf文件为安全模式下的认证文件，需要在FusionInsight Manager中下载principal用户的认证凭证，样例代码中使用的用户为：super，需要修改为准备好的开发用户。
#### 打包项目
- 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考[在Linux环境中调测Spark应用](https://support.huaweicloud.com/devg-lts-mrs/mrs_07_200130.html)。
- 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如"$SPARK_HOME" ）下。
- 将user.keytab、krb5.conf 两个文件上传客户端所在服务器上（文件上传的路径需要和生成的jar包路径一致）。
  ![](https://support.huaweicloud.com/devg-lts-mrs/public_sys-resources/note_3.0-zh-cn.png)
  若运行"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版本（类名等请与实际代码保持一致，此处仅为示例），${ip}请使用实际执行nc -lk 9999的命令的机器ip
  **bin/spark-submit --master yarn --deploy-mode client --class com.huawei.bigdata.spark.examples.**streaming**.JavaHBaseStreamingBulkPutExample SparkOnHbaseJavaExample.jar** **${ip}** **9999 streamingTable** **cf1**
  python版本（文件名等请与实际保持一致，此处仅为示例），假设对应的Java代码打包后包名为**SparkOnHbaseJavaExample.jar**，且放在当前提交目录。
  **bin/spark-submit --master yarn --deploy-mode client --jars** **SparkOnHbaseJavaExample.jar** **HBaseStreamingBulkPutExample.py ${ip} 9999 streamingTable cf1**
  
- yarn-cluster模式： java/scala版本（类名等请与实际代码保持一致，此处仅为示例），${ip}请使用实际执行nc -lk 9999的命令的机器ip
  **bin/spark-submit --master yarn --deploy-mode cluster --class com.huawei.bigdata.spark.examples.**streaming** .JavaHBaseStreamingBulkPutExample **--files /opt/user.keytab,/opt/krb5.conf** SparkOnHbaseJavaExample.jar** **${ip}** **9999 streamingTable** **cf1**
  python版本（文件名等请与实际保持一致，此处仅为示例），假设对应的Java代码打包后包名为**SparkOnHbaseJavaExample.jar**，且放在当前提交目录。
  **bin/spark-submit --master yarn --deploy-mode cluster --files /opt/user.keytab,/opt/krb5.conf** **--jars** **SparkOnHbaseJavaExample.jar** **HBaseStreamingBulkPutExample.py ${ip} 9999 streamingTable cf1**
  
 
#### Java样例代码
下面代码片段仅为演示，具体代码参见SparkOnHbaseJavaExample中JavaHBaseStreamingBulkPutExample文件：
![](https://support.huaweicloud.com/devg-lts-mrs/public_sys-resources/note_3.0-zh-cn.png)
代码中通过awaitTerminationOrTimeout()方法设置了任务超时时间（单位为毫秒），建议根据期望的任务运行时间调整参数大小。
```
public static void main(String[] args) throws IOException {
    if (args.length < 4) {
      System.out.println("JavaHBaseBulkPutExample  " +
              "{host} {port} {tableName}");
      return;
    }
    LoginUtil.loginWithUserKeytab();
    String host = args[0];
    String port = args[1];
    String tableName = args[2];
    String columnFamily = args[3];
    SparkConf sparkConf =
            new SparkConf().setAppName("JavaHBaseStreamingBulkPutExample " +
                    tableName + ":" + port + ":" + tableName);
    JavaSparkContext jsc = new JavaSparkContext(sparkConf);
    try {
      JavaStreamingContext jssc =
              new JavaStreamingContext(jsc, new Duration(1000));
      JavaReceiverInputDStream<String> javaDstream =
              jssc.socketTextStream(host, Integer.parseInt(port));
      Configuration conf = HBaseConfiguration.create();
      JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);
      hbaseContext.streamBulkPut(javaDstream,
              TableName.valueOf(tableName),
              new PutFunction(columnFamily));
      jssc.start();
      jssc.awaitTerminationOrTimeout(60000);
      jssc.stop(false,true);
    }catch(InterruptedException e){
      e.printStackTrace();
    } finally {
      jsc.stop();
    }
  }
```
#### Scala样例代码
下面代码片段仅为演示，具体代码参见SparkOnHbaseScalaExample中HBaseStreamingBulkPutExample文件：
![](https://support.huaweicloud.com/devg-lts-mrs/public_sys-resources/note_3.0-zh-cn.png)
代码中通过awaitTerminationOrTimeout()方法设置了任务超时时间（单位为毫秒），建议根据期望的任务运行时间调整参数大小。
```
def main(args: Array[String]): Unit = {
    loginUtil.loginWithUserKeytab()
    val host = args(0)
    val port = args(1)
    val tableName = args(2)
    val columnFamily = args(3)
    val conf = new SparkConf()
    conf.setAppName("HBase Streaming Bulk Put Example")
    val sc = new SparkContext(conf)
    try {
      val config = HBaseConfiguration.create()
      val hbaseContext = new HBaseContext(sc, config)
      val ssc = new StreamingContext(sc, Seconds(1))
      val lines = ssc.socketTextStream(host, port.toInt)
      hbaseContext.streamBulkPut[String](lines,
        TableName.valueOf(tableName),
        (putRecord) => {
          if (putRecord.length() > 0) {
            val put = new Put(Bytes.toBytes(putRecord))
            put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("foo"), Bytes.toBytes("bar"))
            put
          } else {
            null
          }
        })
      ssc.start()
      ssc.awaitTerminationOrTimeout(60000)
      ssc.stop(stopSparkContext = false)
    } finally {
      sc.stop()
    }
  }
```
#### Python样例代码
下面代码片段仅为演示，具体代码参见SparkOnHbasePythonExample中HBaseStreamingBulkPutExample文件：
```
# -*- coding:utf-8 -*-
"""
【说明】
(1)由于pyspark不提供Hbase相关api,本样例使用Python调用Java的方式实现
(2)如果使用yarn-client模式运行,请确认Spark2x客户端Spark2x/spark/conf/spark-defaults.conf中
   spark.yarn.security.credentials.hbase.enabled参数配置为true
"""
from py4j.java_gateway import java_import
from pyspark.sql import SparkSession
# 创建SparkSession
spark = SparkSession\
        .builder\
        .appName("JavaHBaseStreamingBulkPutExample")\
        .getOrCreate()
# 向sc._jvm中导入要运行的类
java_import(spark._jvm, 'com.huawei.bigdata.spark.examples.streaming.JavaHBaseStreamingBulkPutExample')
# 创建类实例并调用方法，传递sc._jsc参数
spark._jvm.JavaHBaseStreamingBulkPutExample().execute(spark._jsc, sys.argv)
# 停止SparkSession
spark.stop()
```