批量加载-Bulk Load
在工作过程中有个需求，需要将DataFrame的数据保存进Hbase,并且在Spark集群并没有安装Hbase,此时对于常规的使用put将DataFrame加载进Hbase的方式不在适用，一方面是没有Hbase,另一方面是数据量比较大，通过Put加载数据太慢。
为了实现自己的需求，测试了两个方案
通过hive关联Hbase将数据加载进去
大致步骤
在之前已经有一篇博文是关于hive关联Hbase的，再次仅仅简述处理过程
1.首先将DataFrame进行处理后保存成文件resultFile
2.根据DataFrame字段建立一张Hive临时表tmp_table1
3.将resultFile通过load data 导入临时表tmp_table1
4.建立一张关联Hbase的表hive_hbase_table
5.通过insert into 将tmp_table1的数据插入hive_hbase_table
此时，实现了将DataFrame存入Hbase的目的，如果Spark与Hbase不在同一个集群，那么两个集群之间只需要传输hive底层文件即可。
此方式的优点和缺点
优点：
1.直接保存为hive底层文件(即DataFrame直接写文件，对于里面的字段不需要过多处理)
2.由于hive表关联了Hbase，我们可以通过hive直接查询Hbase的数据(方便对数据进行二次加工)
缺点：
1.需要进行两次数据加载：
第一次加载是根据因为是load速度还很快，基本可以忽略不计；
但是第二第insert into 会触发mapreduce，执行时间与数据量的大小关系很大。
2.耗费额外的空间：
临时表tmp_table1仅仅是作为一个中转站，但是还是会消耗磁盘空间，这种方式消耗磁盘空间的大小接近最终结果的两倍。
具体参考此篇文章
通过HFile 直接将数据导入Hbase
由于客户对数据导入的速度有一定要求，所以第一个方案成为备选方案。于是在查询资料的情况下，找到了第二个方案。
将DataFrame先保存成HFile,然后通过命令直接将文件HFile load进对应的Hbase表。
接下来将详细记录我在这个过程中的踩坑，脱坑历程。
1.第一坑：包找不到
参考了众多博文（其实都是一样的），有几个方案本想来参考一下的，但是自己用的时候发现有些包根本就找不到。于是又花了点时间去看看对应的类在哪个包，依赖是啥。
在此处给出我所使用的相关包依赖（sbt的，maven对应改一下基本上就可以了）
//hbase的几个包必须要有，否则有些关键的类用不了
libraryDependencies += "org.apache.spark" % "spark-core_2.10" % "1.6.0-cdh5.7.2"
libraryDependencies += "org.apache.spark" % "spark-sql_2.10" % "1.6.0-cdh5.7.2"
libraryDependencies += "org.apache.spark" % "spark-hive_2.10" % "1.6.0-cdh5.7.2"
libraryDependencies += "org.apache.hbase" % "hbase-client" % "1.2.0-cdh5.7.2"
libraryDependencies += "org.apache.hbase" % "hbase-server" % "1.2.0-cdh5.7.2"
libraryDependencies += "org.apache.hbase" % "hbase-common" % "1.2.0-cdh5.7.2"
libraryDependencies += "org.apache.hbase" % "hbase-protocol" % "1.2.0-cdh5.7.2"
2.第二坑：需不需要配置Hbase参数？
因为我们有三个环境，开发环境是有Hbase的，生产没有，而最终数据又要放到第三个集群，所以Hbase 的有无对结果的影响还是比较纠结。
经过测试，如果单纯的在Spark里面生成HFile，则不需要配置额外的信息，直接默认即可，
但是如果在Spark生成HFile向直接导入Hbase则需要配置相关信息。详细的情况后续说明。
3.第三坑：HFile在代码里面load不进去？
一开始我想的是在生成HFile后直接导入Hbase(单纯想测试的)，可是HFile死活load不进去，一直出现一下的信息:
18/10/13 15:12:41 INFO mapreduce.LoadIncrementalHFiles: Trying to load hfile=hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile/cf_info/5eab4cb95552481faf538552f848b038 first=752-\xE6\x83\xA0\xE5\xB7\x9E last=752-\xE6\x83\xA0\xE5\xB7\x9E
18/10/13 15:12:41 INFO mapreduce.LoadIncrementalHFiles: Trying to load hfile=hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile/cf_info/5b39a5e9779e499bb6817affc94c51a9 first=662-\xE9\x98\xB3\xE6\xB1\x9F last=662-\xE9\x98\xB3\xE6\xB1\x9F
18/10/13 15:12:41 INFO mapreduce.LoadIncrementalHFiles: Trying to load hfile=hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile/cf_info/2e3334a0a8094b899fb6cef2195becd5 first=751-\xE9\x9F\xB6\xE5\x85\xB3 last=751-\xE9\x9F\xB6\xE5\x85\xB3
18/10/13 15:12:41 INFO mapreduce.LoadIncrementalHFiles: Trying to load hfile=hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile/cf_info/e1ebb7ab430e47dfbc05938e037c00de first=759-\xE6\xB9\x9B\xE6\xB1\x9F last=759-\xE6\xB9\x9B\xE6\xB1\x9F
18/10/13 15:23:55 INFO client.RpcRetryingCaller: Call exception, tries=10, retries=35, started=673838 ms ago, cancelled=false, msg=row '' on table 'iptv:spark_test' at region=iptv:spark_test,,1539413066165.047ff19285b4360a8cf82bdade12a465., hostname=iptve2e06,60020,1539323044523, seqNum=2
18/10/13 15:25:10 INFO client.RpcRetryingCaller: Call exception, tries=11, retries=35, started=748954 ms ago, cancelled=false, msg=row '' on table 'iptv:spark_test' at region=iptv:spark_test,,1539413066165.047ff19285b4360a8cf82bdade12a465., hostname=iptve2e06,60020,1539323044523, seqNum=2
找了很久也没有找到解决方法（我猜测可能是读写权限问题，因为我读取数据是没问题的），最后只能在外面执行命令导入数据了。
虽然这个直接在代码处理是不符合我的需求的，但是这个问题没搞定很不爽（后期找到解决方案在发出来）
4.第四坑：只能处理一列数据
我的DataFrame最终的存储目标是一个列族，多个列的，但是目前在网上和官网的都是只能处理单个列族，单个列，与我自己的需求相差甚远。在网上找到一个方案可以实现一列族、多列但是需要修改源代码，这个由于时间关系就放弃。
最后采取一个折中的办法，就是每次处理一列最后合并一下，很蠢的办法，但是目前只想到这个。
在我的另一篇博文Spark：DataFrame写HFile （Hbase）一个列族、一个列扩展一个列族、多个列中实现了多列处理
5.第五坑：用错导入数据的命令
在直接在代码load数据失败后，就寻求其他的办法来load数据
在看了hbase的官方文档后，知道还可以在外部用hbase的命令将数据load进去
//在Hbase官方文档找到的命令
hbase org.apache.hadoop.hbase.tool.LoadIncrementalHFiles hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile iptv:spark_test
//在某篇博文里面找到的命令
hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile iptv:spark_test
当时用了官方文档的命令 直接报找不到包，整个人瞬间不好了。
后面再去搜相关资料看到别人用的命令是下面这个，试了一下居然成功了，在Hbase也能查到对应的数据了，至此，才算是完成了整个流程测试。
实现代码
1.直接在代码里面处理load数据流程
注：这个最后一步我没有走通，问题还在找。
package com.iptv.job.basedata
import java.text.SimpleDateFormat
import java.util.{Calendar, Date}
import com.iptv.domain.DatePattern
import com.iptv.job.JobBase
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.{HBaseConfiguration, KeyValue, TableName}
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
/**
* @author 利伊奥克儿-lillcol
* 2018/10/12-15:58
*
*/
object HbaseHFileTest {
var hdfsPath: String = ""
var proPath: String = ""
var DATE: String = ""
var conf: Configuration = null
val sparkConf: SparkConf = new SparkConf().setAppName(getClass.getSimpleName)
val sc: SparkContext = new SparkContext(sparkConf)
val sqlContext: SQLContext = getSQLContext(sc)
import sqlContext.implicits._
def main(args: Array[String]): Unit = {
val date: Date = getCommandParam(args)
hdfsPath = args(0)
proPath = args(1)
DATE = dateHelper.getDateFormatStr(date, DatePattern.DATE_YMD)
//获取测试DataFrame
val dim_sys_city_dict: DataFrame = readMysqlTable(sqlContext, "DIM_SYS_CITY_DICT", proPath)
val resultDataFrame: DataFrame = dim_sys_city_dict
.select(concat($"city_id", lit("-"), $"city_name").as("key"), $"city_id", $"city_name")
//hbase的表名
val tableName = "iptv:spark_test"
//hbase的参数配置
conf = HBaseConfiguration.create()
conf.set(TableOutputFormat.OUTPUT_TABLE, tableName)
conf.set("hbase.zookeeper.quorum", "n1,n2,n3")
conf.set("hbase.zookeeper.property.clientPort", "2181")
conf.set("hbase.zookeeper.master", "n1:60000")
conf.set("hbase.zookeeper.master", "n1:60000")
lazy val job = Job.getInstance(conf)
val con: Connection = ConnectionFactory.createConnection(conf)
//设置MapOutput Key Value 的数据类型
job.setMapOutputKeyClass(classOf[ImmutableBytesWritable])
job.setMapOutputValueClass(classOf[KeyValue])
// val table: Table =new HTable(conf,tableName) //这个方法被 deprecated 采用下面的方式
val table: Table = con.getTable(TableName.valueOf(tableName))
val cf: Array[Byte] = "cf_info".getBytes //列族
//对第一个列处理
val result1: RDD[(ImmutableBytesWritable, KeyValue)] = resultDataFrame
.map(row => {
//将rdd转换成HFile需要的格式,我们上面定义了Hfile的key是ImmutableBytesWritable,那么我们定义的RDD也是要以ImmutableBytesWritable的实例为key
val rowkey = Bytes.toBytes(row.getAs[String]("key")) //key
val clounmVale: Array[Byte] = "city_id".getBytes //列的名称
val value: Array[Byte] = Bytes.toBytes(row.getAs[String]("city_id")) //列的值
val kv: KeyValue = new KeyValue(rowkey, cf, clounmVale, value) //封装一下 rowkey, cf, clounmVale, value
(new ImmutableBytesWritable(rowkey), kv) //生成 KeyValue 这里每次只能处理一列（后续看看有没办法处理多列）
})
//对第而个列处理
val result2: RDD[(ImmutableBytesWritable, KeyValue)] = resultDataFrame
.map(row => {
//将rdd转换成HFile需要的格式,我们上面定义了Hfile的key是ImmutableBytesWritable,那么我们定义的RDD也是要以ImmutableBytesWritable的实例为key
val rowkey = Bytes.toBytes(row.getAs[String]("key")) //key
val clounmVale: Array[Byte] = "city_name".getBytes //列的名称
val value: Array[Byte] = Bytes.toBytes(row.getAs[String]("city_name")) //列的值
val kv: KeyValue = new KeyValue(rowkey, cf, clounmVale, value) //封装一下 rowkey, cf, clounmVale, value
(new ImmutableBytesWritable(rowkey), kv) //生成 KeyValue 这里每次只能处理一列（后续看看有没办法处理多列）
})
//保存 到hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile"
//union两个数据
result1
.union(result2)
.sortBy(x => x._1, true) //要保持 key 整理有序
.saveAsNewAPIHadoopFile("hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile",
classOf[ImmutableBytesWritable],
classOf[KeyValue],
classOf[HFileOutputFormat2],
job.getConfiguration)
// 将保存在临时文件夹的hfile数据保存到hbase中
val load = new LoadIncrementalHFiles(conf)
load.doBulkLoad(new Path("hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile"), table.asInstanceOf[HTable])
}
}
2.在外面通过命令load数据
package com.iptv.job.basedata
import java.text.SimpleDateFormat
import java.util.{Calendar, Date}
import com.iptv.domain.DatePattern
import com.iptv.job.JobBase
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.{HBaseConfiguration, KeyValue, TableName}
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
/**
* @author 利伊奥克儿-lillcol
* 2018/10/12-15:58
*
*/
object HbaseHFileTest {
var hdfsPath: String = ""
var proPath: String = ""
var DATE: String = ""
var conf: Configuration = null
val sparkConf: SparkConf = new SparkConf().setAppName(getClass.getSimpleName)
val sc: SparkContext = new SparkContext(sparkConf)
val sqlContext: SQLContext = getSQLContext(sc)
import sqlContext.implicits._
def main(args: Array[String]): Unit = {
val date: Date = getCommandParam(args)
hdfsPath = args(0)
proPath = args(1)
DATE = dateHelper.getDateFormatStr(date, DatePattern.DATE_YMD)
//获取测试DataFrame
val dim_sys_city_dict: DataFrame = readMysqlTable(sqlContext, "DIM_SYS_CITY_DICT", proPath)
val resultDataFrame: DataFrame = dim_sys_city_dict
.select(concat($"city_id", lit("-"), $"city_name").as("key"), $"city_id", $"city_name")
//hbase的表名
val tableName = "iptv:spark_test"
//hbase的参数配置
conf = HBaseConfiguration.create()
conf.set(TableOutputFormat.OUTPUT_TABLE, tableName)
lazy val job = Job.getInstance(conf)
val con: Connection = ConnectionFactory.createConnection(conf)
//设置MapOutput Key Value 的数据类型
job.setMapOutputKeyClass(classOf[ImmutableBytesWritable])
job.setMapOutputValueClass(classOf[KeyValue])
// val table: Table =new HTable(conf,tableName) //这个方法被 deprecated 采用下面的方式
val table: Table = con.getTable(TableName.valueOf(tableName))
val cf: Array[Byte] = "cf_info".getBytes //列族
//对第一个列处理
val result1: RDD[(ImmutableBytesWritable, KeyValue)] = resultDataFrame
.map(row => {
//将rdd转换成HFile需要的格式,我们上面定义了Hfile的key是ImmutableBytesWritable,那么我们定义的RDD也是要以ImmutableBytesWritable的实例为key
val rowkey = Bytes.toBytes(row.getAs[String]("key")) //key
val clounmVale: Array[Byte] = "city_id".getBytes //列的名称
val value: Array[Byte] = Bytes.toBytes(row.getAs[String]("city_id")) //列的值
val kv: KeyValue = new KeyValue(rowkey, cf, clounmVale, value) //封装一下 rowkey, cf, clounmVale, value
(new ImmutableBytesWritable(rowkey), kv) //生成 KeyValue 这里每次只能处理一列（后续看看有没办法处理多列）
})
//对第而个列处理
val result2: RDD[(ImmutableBytesWritable, KeyValue)] = resultDataFrame
.map(row => {
//将rdd转换成HFile需要的格式,我们上面定义了Hfile的key是ImmutableBytesWritable,那么我们定义的RDD也是要以ImmutableBytesWritable的实例为key
val rowkey = Bytes.toBytes(row.getAs[String]("key")) //key
val clounmVale: Array[Byte] = "city_name".getBytes //列的名称
val value: Array[Byte] = Bytes.toBytes(row.getAs[String]("city_name")) //列的值
val kv: KeyValue = new KeyValue(rowkey, cf, clounmVale, value) //封装一下 rowkey, cf, clounmVale, value
(new ImmutableBytesWritable(rowkey), kv) //生成 KeyValue 这里每次只能处理一列（后续看看有没办法处理多列）
})
//s"保存 到hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile")
//union两个数据
result1
.union(result2)
.sortBy(x => x._1, true) //要保持 key 整理有序
.saveAsNewAPIHadoopFile("hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile",
classOf[ImmutableBytesWritable],
classOf[KeyValue],
classOf[HFileOutputFormat2],
job.getConfiguration)
}
}
//通过下面的命令将数据loadhbase
hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile iptv:spark_test
//生成的HFile
[hdfs@iptve2e03 tmp_lillcol]$ hadoop fs -du -h hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile
0 0 hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile/_SUCCESS
20.3 K 40.6 K hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile/cf_info
//数据load 进hbase后
[hdfs@iptve2e03 tmp_lillcol]$ hadoop fs -du -h hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile
0 0 hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile/_SUCCESS
0 0 hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile/cf_info
//在hbase下的文件结构
[hdfs@iptve2e03 tmp_lillcol]$ hadoop fs -du -h hdfs://ns1/hbase/data/iptv/spark_test
290 580 hdfs://ns1/hbase/data/iptv/spark_test/.tabledesc
0 0 hdfs://ns1/hbase/data/iptv/spark_test/.tmp
32.4 K 64.8 K hdfs://ns1/hbase/data/iptv/spark_test/047ff19285b4360a8cf82bdade12a465
//Hbase 数据情况
hbase(main):005:0> scan 'iptv:spark_test' ,{LIMIT=>5}
ROW COLUMN+CELL
200-\xE5\xB9\xBF\xE5\xB7\x9E column=cf_info:city_id, timestamp=1539496131471, value=200
200-\xE5\xB9\xBF\xE5\xB7\x9E column=cf_info:city_name, timestamp=1539496131471, value=\xE5\xB9\xBF\xE5\xB7\x9E
660-\xE6\xB1\x95\xE5\xB0\xBE column=cf_info:city_id, timestamp=1539496131471, value=660
660-\xE6\xB1\x95\xE5\xB0\xBE column=cf_info:city_name, timestamp=1539496131471, value=\xE6\xB1\x95\xE5\xB0\xBE
662-\xE9\x98\xB3\xE6\xB1\x9F column=cf_info:city_id, timestamp=1539496131471, value=662
662-\xE9\x98\xB3\xE6\xB1\x9F column=cf_info:city_name, timestamp=1539496131471, value=\xE9\x98\xB3\xE6\xB1\x9F
663-\xE6\x8F\xAD\xE9\x98\xB3 column=cf_info:city_id, timestamp=1539496131471, value=663
663-\xE6\x8F\xAD\xE9\x98\xB3 column=cf_info:city_name, timestamp=1539496131471, value=\xE6\x8F\xAD\xE9\x98\xB3
668-\xE8\x8C\x82\xE5\x90\x8D column=cf_info:city_id, timestamp=1539496131994, value=668
668-\xE8\x8C\x82\xE5\x90\x8D column=cf_info:city_name, timestamp=1539496131994, value=\xE8\x8C\x82\xE5\x90\x8D
5 row(s) in 0.0590 seconds
hbase(main):006:0>
至此，将数据通过HFile批量导入到Hbase的目的实现
通过生成HFilede的方式将load Hbase的方式速度是真的快，给人感觉是仅仅是移动了文件
刚接触Hbase 发现还是有点意思的
