用戶畫像-1: 大批量資料匯入HBase

知識拓展：如何使用SQOOP進行增量匯入資料至HBase表，范例命令如下：

例一:

/export/servers/sqoop/bin/sqoop import \
-D sqoop.hbase.add.row.key=true \
--connect jdbc:mysql://bigdata-cdh01.itcast.cn:3306/tags_dat \
--username root \
--password 123456 \
--table tbl_users \
--hbase-create-table \
--hbase-table tbl_users \
--column-family detail \
--hbase-row-key id \
--num-mappers 2

/export/servers/sqoop/bin/sqoop import \
-D sqoop.hbase.add.row.key=true \
--connect jdbc:mysql://bigdata-cdh01.itcast.cn:3306/tags_dat \
--username root \
--password 123456 \
--table tbl_logs \
--hbase-create-table \
--hbase-table tag_logs \
--column-family detail \
--hbase-row-key id \
--num-mappers 20 \
--incremental lastmodified \
--check-column log_time \

相關增量匯入引數說明：

HADOOP_HOME=/export/servers/hadoop 
HBASE_HOME=/export/servers/hbase 
HADOOP_CLASSPATH=`${HBASE_HOME}/bin/hbase mapredcp`:${HBASE_HOME}/conf ${HADOOP_HOME}/bin/yarn jar ${HBASE_HOME}/lib/hbase-server-1.2.0- cdh5.14.0.jar

An example program must be given as the first argument.
Valid program names are:
CellCounter: Count cells in HBase table.
WALPlayer: Replay WAL files.
completebulkload: Complete a bulk data load.
copytable: Export a table from local cluster to peer cluster.
export: Write table data to HDFS.
exportsnapshot: Export the specific snapshot to a given FileSystem.
import: Import data written by Export.
importtsv: Import data in TSV format.
rowcounter: Count rows in HBase table.
verifyrep: Compare the data from tables in two different clusters.

Usage: importtsv -Dimporttsv.columns=a,b,c <tablename> <inputdir>
The column names of the TSV data must be specified using the -
Dimporttsv.columns
option. This option takes the form of comma-separated column names, where
each
column name is either a simple column family, or a columnfamily:qualifier.
The special column name HBASE_ROW_KEY is used to designate that this column
should be used as the row key for each imported record.
To instead generate HFiles of data to prepare for a bulk data load, pass
the option:
-Dimporttsv.bulk.output=/path/for/output
'-Dimporttsv.separator=|' - eg separate on pipes instead of tabs
For performance consider the following options:
-Dmapreduce.map.speculative=false
-Dmapreduce.reduce.speculative=false

HADOOP_HOME=/export/servers/hadoop
HBASE_HOME=/export/servers/hbase
HADOOP_CLASSPATH=`${HBASE_HOME}/bin/hbase mapredcp`:${HBASE_HOME}/conf
${HADOOP_HOME}/bin/yarn jar ${HBASE_HOME}/lib/hbase-server-1.2.0-
cdh5.14.0.jar \
importtsv \
-Dimporttsv.columns=HBASE_ROW_KEY,detail:log_id,detail:remote_ip,detail:s
ite_global_ticket,detail:site_global_session,detail:global_user_id,detai
l:cookie_text,detail:user_agent,detail:ref_url,detail:loc_url,detail:log
_time \
tbl_logs \
/user/hive/warehouse/tags_dat.db/tbl_logs

回顧一下：
采用Put方式向HBase表中插入資料流程：
Put
-> WAL 預寫日志
-> MemStore(記憶體) ，當達到一定大寫Spill到磁盤上：
StoreFile（
HFile）
思考：
對海量資料插入，能否將資料直接保存為HFile檔案，然后加載到HBase表中

# 1. 生成HFILES檔案
HADOOP_HOME=/export/servers/hadoop
HBASE_HOME=/export/servers/hbase
HADOOP_CLASSPATH=`${HBASE_HOME}/bin/hbase mapredcp`:${HBASE_HOME}/conf
${HADOOP_HOME}/bin/yarn jar ${HBASE_HOME}/lib/hbase-server-1.2.0-
cdh5.14.0.jar \
importtsv \
-Dimporttsv.bulk.output=hdfs://bigdata-
cdh01.itcast.cn:8020/datas/output_hfile/tbl_logs \
-
Dimporttsv.columns=HBASE_ROW_KEY,detail:log_id,detail:remote_ip,detail:
site_global_ticket,detail:site_global_session,detail:global_user_id,det
ail:cookie_text,detail:user_agent,detail:ref_url,detail:loc_url,detail:
log_time \

tbl_logs \ 
/user/hive/warehouse/tags_dat.db/tbl_logs
 # 2. 將HFILE檔案加載到表中 
HADOOP_CLASSPATH=`${HBASE_HOME}/bin/hbase mapredcp`:${HBASE_HOME}/conf ${HADOOP_HOME}/bin/yarn jar ${HBASE_HOME}/lib/hbase-server-1.2.0- cdh5.14.0.jar \ completebulkload \ 
hdfs://bigdata-cdh01.itcast.cn:8020/datas/output_hfile/tbl_logs \ tbl_logs

val put = new Put(rowKeyByts) 
put.addColumn(cf, column, value) 
put.addColumn(cf, column, value) 
put.addColumn(cf, column, value) 
put.addColumn(cf, column, value) 
table.put(put)

1）、Extract，異構資料源資料匯入到 HDFS 之上， 
2）、Transform，通過用戶代碼，可以是 MR 或者 Spark 任務將資料轉化為 HFile， 
3）、Load，HFile 通過 loadIncrementalHFiles 呼叫將 HFile 放置到 Region 對應的 HDFS 目錄上，該程序可能涉及到檔案切分， 
1、不會觸發WAL預寫日志，當表還沒有資料時進行資料匯入不會產生Flush和Split， 
2、減少介面呼叫的消耗，是一種快速寫入的優化方式，
 Spark讀寫HBase之使用Spark自帶的API以及使用Bulk Load將大量資料匯入HBase： https://www.jianshu.com/p/b6c5a5ba30af

Bulkload程序主要包括三部分：
1、從資料源(通常是文本檔案或其他的資料庫)提取資料并上傳到HDFS，
抽取資料到HDFS和Hbase并沒有關系，所以大家可以選用自己擅長的方式進行，
2、利用MapReduce作業處理事先準備的資料 ，
這一步需要一個MapReduce作業，并且大多數情況下還需要我們自己撰寫Map函式，而Reduce
函式不需要我們考慮，由HBase提供，
該作業需要使用rowkey(行鍵)作為輸出Key；KeyValue、Put或者Delete作為輸出Value，
MapReduce作業需要使用HFileOutputFormat2來生成HBase資料檔案，
為了有效的匯入資料，需要配置HFileOutputFormat2使得每一個輸出檔案都在一個合適的區
域中，為了達到這個目的，MapReduce作業會使用Hadoop的TotalOrderPartitioner類根據表的
key值將輸出分割開來，
HFileOutputFormat2的方法configureIncrementalLoad()會自動的完成上面的作業，
3、告訴RegionServers資料的位置并匯入資料，
這一步是最簡單的，通常需要使用LoadIncrementalHFiles(更為人所熟知是
completebulkload工具)，將檔案在HDFS上的位置傳遞給它，它就會利用RegionServer將資料導
入到相應的區域，