大數據查詢——HBase讀寫設計與實踐--轉

背景介紹

本項目主要解決 check 和 opinion2 張歷史數據表（歷史數據是指當業務發生過程中的完整中間流程和結果數據）的在線查詢。原實現基於 Oracle 提供存儲查詢服務，隨着數據量的不斷增加，在寫入和讀取過程中面臨性能問題，且歷史數據僅供業務查詢參考，並不影響實際流程，從系統結構上來說，放在業務鏈條上游比較重。本項目將其置於下游數據處理 Hadoop 分布式平台來實現此需求。下面列一些具體的需求指標：

1. 數據量：目前 check 表的累計數據量為 5000w+ 行，11GB；opinion 表的累計數據量為 3 億 +，約 100GB。每日增量約為每張表 50 萬 + 行，只做 insert，不做 update。
2. 查詢要求：check 表的主鍵為 id（Oracle 全局 id），查詢鍵為 check_id，一個 check_id 對應多條記錄，所以需返回對應記錄的 list； opinion 表的主鍵也是 id，查詢鍵是 bussiness_no 和 buss_type，同理返回 list。單筆查詢返回 List 大小約 50 條以下，查詢頻率為 100 筆 / 天左右，查詢響應時間 2s。

技術選型

從數據量及查詢要求來看，分布式平台上具備大數據量存儲，且提供實時查詢能力的組件首選 HBase。根據需求做了初步的調研和評估后，大致確定 HBase 作為主要存儲組件。將需求拆解為寫入和讀取 HBase 兩部分。

讀取 HBase 相對來說方案比較確定，基本根據需求設計 RowKey，然后根據 HBase 提供的豐富 API（get，scan 等）來讀取數據，滿足性能要求即可。

寫入 HBase 的方法大致有以下幾種：

1. Java 調用 HBase 原生 API，HTable.add(List(Put))。
2. MapReduce 作業，使用 TableOutputFormat 作為輸出。
3. Bulk Load，先將數據按照 HBase 的內部數據格式生成持久化的 HFile 文件，然后復制到合適的位置並通知 RegionServer ，即完成海量數據的入庫。其中生成 Hfile 這一步可以選擇 MapReduce 或 Spark。

本文采用第 3 種方式，Spark + Bulk Load 寫入 HBase。該方法相對其他 2 種方式有以下優勢：

1. BulkLoad 不會寫 WAL，也不會產生 flush 以及 split。
2. 如果我們大量調用 PUT 接口插入數據，可能會導致大量的 GC 操作。除了影響性能之外，嚴重時甚至可能會對 HBase 節點的穩定性造成影響，采用 BulkLoad 無此顧慮。
3. 過程中沒有大量的接口調用消耗性能。
4. 可以利用 Spark 強大的計算能力。

圖示如下：

設計

環境信息

Hadoop 2.5-2.7
HBase 0.98.6
Spark 2.0.0-2.1.1
Sqoop 1.4.6

表設計

本段的重點在於討論 HBase 表的設計，其中 RowKey 是最重要的部分。為了方便說明問題，我們先來看看數據格式。以下以 check 舉例，opinion 同理。

check 表（原表字段有 18 個，為方便描述，本文截選 5 個字段示意）

如上圖所示，主鍵為 id，32 位字母和數字隨機組成，業務查詢字段 check_id 為不定長字段（不超過 32 位），字母和數字組成，同一 check_id 可能對應多條記錄，其他為相關業務字段。眾所周知，HBase 是基於 RowKey 提供查詢，且要求 RowKey 是唯一的。RowKey 的設計主要考慮的是數據將怎樣被訪問。初步來看，我們有 2 種設計方法。

1. 拆成 2 張表，一張表 id 作為 RowKey，列為 check 表對應的各列；另一張表為索引表，RowKey 為 check_id，每一列對應一個 id。查詢時，先找到 check_id 對應的 id list，然后根據 id 找到對應的記錄。均為 HBase 的 get 操作。
2. 將本需求可看成是一個范圍查詢，而不是單條查詢。將 check_id 作為 RowKey 的前綴，后面跟 id。查詢時設置 Scan 的 startRow 和 stopRow，找到對應的記錄 list。

第一種方法優點是表結構簡單，RowKey 容易設計，缺點為 1）數據寫入時，一行原始數據需要寫入到 2 張表，且索引表寫入前需要先掃描該 RowKey 是否存在，如果存在，則加入一列，否則新建一行，2）讀取的時候，即便是采用 List, 也至少需要讀取 2 次表。第二種設計方法，RowKey 設計較為復雜，但是寫入和讀取都是一次性的。綜合考慮，我們采用第二種設計方法。

RowKey 設計

熱點問題

HBase 中的行是以 RowKey 的字典序排序的，其熱點問題通常發生在大量的客戶端直接訪問集群的一個或極少數節點。默認情況下，在開始建表時，表只會有一個 region，並隨着 region 增大而拆分成更多的 region，這些 region 才能分布在多個 regionserver 上從而使負載均分。對於我們的業務需求，存量數據已經較大，因此有必要在一開始就將 HBase 的負載均攤到每個 regionserver，即做 pre-split。常見的防治熱點的方法為加鹽，hash 散列，自增部分（如時間戳）翻轉等。

RowKey 設計

Step1：確定預分區數目，創建 HBase Table

不同的業務場景及數據特點確定數目的方式不一樣，我個人認為應該綜合考慮數據量大小和集群大小等因素。比如 check 表大小約為 11G，測試集群大小為 10 台機器，hbase.hregion.max.filesize=3G（當 region 的大小超過這個數時，將拆分為 2 個），所以初始化時盡量使得一個 region 的大小為 1~2G（不會一上來就 split），region 數據分到 11G/2G=6 個，但為了充分利用集群資源，本文中 check 表划分為 10 個分區。如果數據量為 100G，且不斷增長，集群情況不變，則 region 數目增大到 100G/2G=50 個左右較合適。Hbase check 表建表語句如下：

create 'tinawang:check', { NAME => 'f', COMPRESSION => 'SNAPPY',DATA_BLOCK_ENCODING => 'FAST_DIFF',BLOOMFILTER=>'ROW'}, {SPLITS => [ '1','2','3', '4','5','6','7','8','9']}

其中，Column Family =‘f’，越短越好。

COMPRESSION => 'SNAPPY'，HBase 支持 3 種壓縮 LZO, GZIP and Snappy。GZIP 壓縮率高，但是耗 CPU。后兩者差不多，Snappy 稍微勝出一點，cpu 消耗的比 GZIP 少。一般在 IO 和 CPU 均衡下，選擇 Snappy。

DATA_BLOCK_ENCODING => 'FAST_DIFF'，本案例中 RowKey 較為接近，通過以下命令查看 key 長度相對 value 較長。

./hbase org.apache.hadoop.hbase.io.hfile.HFile -m -f /apps/hbase/data/data/tinawang/check/a661f0f95598662a53b3d8b1ae469fdf/f/a5fefc880f87492d908672e1634f2eed_SeqId_2_

Step2：RowKey 組成

Salt

讓數據均衡的分布到各個 Region 上，結合 pre-split，我們對查詢鍵即 check 表的 check_id 求 hashcode 值，然后 modulus(numRegions) 作為前綴，注意補齊數據。

StringUtils.leftPad(Integer.toString(Math.abs(check_id.hashCode() % numRegion)),1,’0’) 

說明：如果數據量達上百 G 以上，則 numRegions 自然到 2 位數，則 salt 也為 2 位。

Hash 散列

因為 check_id 本身是不定長的字符數字串，為使數據散列化，方便 RowKey 查詢和比較，我們對 check_id 采用 SHA1 散列化，並使之 32 位定長化。

MD5Hash.getMD5AsHex(Bytes.toBytes(check_id))

唯一性

以上 salt+hash 作為 RowKey 前綴，加上 check 表的主鍵 id 來保障 RowKey 唯一性。綜上，check 表的 RowKey 設計如下：（check_id=A208849559）

為增強可讀性，中間還可以加上自定義的分割符，如’+’,’|’等。

7+7c9498b4a83974da56b252122b9752bf+56B63AB98C2E00B4E053C501380709AD

以上設計能保證對每次查詢而言，其 salt+hash 前綴值是確定的，並且落在同一個 region 中。需要說明的是 HBase 中 check 表的各列同數據源 Oracle 中 check 表的各列存儲。

WEB 查詢設計

RowKey 設計與查詢息息相關，查詢方式決定 RowKey 設計，反之基於以上 RowKey 設計，查詢時通過設置 Scan 的 [startRow，stopRow], 即可完成掃描。以查詢 check_id=A208849559 為例，根據 RowKey 的設計原則，對其進行 salt+hash 計算，得前綴。

startRow = 7+7c9498b4a83974da56b252122b9752bf stopRow = 7+7c9498b4a83974da56b252122b9752bg

代碼實現關鍵流程

Spark write to HBase

Step0: prepare work

因為是從上游系統承接的業務數據，存量數據采用 sqoop 抽到 hdfs；增量數據每日以文件的形式從 ftp 站點獲取。因為業務數據字段中包含一些換行符，且 sqoop1.4.6 目前只支持單字節，所以本文選擇’0x01’作為列分隔符，’0x10’作為行分隔符。

Step1: Spark read hdfs text file

SparkContext.textfile() 默認行分隔符為”\n”，此處我們用“0x10”，需要在 Configuration 中配置。應用配置，我們調用 newAPIHadoopFile 方法來讀取 hdfs 文件，返回 JavaPairRDD，其中 LongWritable 和 Text 分別為 Hadoop 中的 Long 類型和 String 類型（所有 Hadoop 數據類型和 java 的數據類型都很相像，除了它們是針對網絡序列化而做的特殊優化）。我們需要的數據文件放在 pairRDD 的 value 中，即 Text 指代。為后續處理方便，可將 JavaPairRDD轉換為 JavaRDD< String >。

Step2: Transfer and sort RDD

① 將 avaRDD< String>轉換成 JavaPairRDD<tuple2,String>，其中參數依次表示為，RowKey，col，value。做這樣轉換是因為 HBase 的基本原理是基於 RowKey 排序的，並且當采用 bulk load 方式將數據寫入多個預分區（region）時，要求 Spark 各 partition 的數據是有序的，RowKey，column family（cf），col name 均需要有序。在本案例中因為只有一個列簇，所以將 RowKey 和 col name 組織出來為 Tuple2格式的 key。請注意原本數據庫中的一行記錄（n 個字段），此時會被拆成 n 行。

② 基於 JavaPairRDD<tuple2,String>進行 RowKey，col 的二次排序。如果不做排序，會報以下異常：

java.io.IOException: Added a key notlexically larger than previous key

③ 將數據組織成 HFile 要求的 JavaPairRDDhfileRDD。

Step3：create hfile and bulk load to HBase

①主要調用 saveAsNewAPIHadoopFile 方法：

hfileRdd.saveAsNewAPIHadoopFile(hfilePath,ImmutableBytesWritable.class,
KeyValue.class,HFileOutputFormat2.class,config);

② hfilebulk load to HBase

final Job job = Job.getInstance(); job.setMapOutputKeyClass(ImmutableBytesWritable.class); job.setMapOutputValueClass(KeyValue.class); HFileOutputFormat2.configureIncrementalLoad(job,htable); LoadIncrementalHFiles bulkLoader = newLoadIncrementalHFiles(config); bulkLoader.doBulkLoad(newPath(hfilePath),htable);

注：如果集群開啟了 kerberos，step4 需要放置在 ugi.doAs（）方法中，在進行如下驗證后實現

UserGroupInformation ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(keyUser,keytabPath);
UserGroupInformation.setLoginUser(ugi);

訪問 HBase 集群的 60010 端口 web，可以看到 region 分布情況。

Read from HBase

本文基於 spring boot 框架來開發 web 端訪問 HBase 內數據。

use connection pool(使用連接池)

創建連接是一個比較重的操作，在實際 HBase 工程中，我們引入連接池來共享 zk 連接，meta 信息緩存，region server 和 master 的連接。

HConnection connection = HConnectionManager.createConnection(config); HTableInterface table = connection.getTable("table1"); try { // Use the table as needed, for a single operation and a single thread } finally { table.close(); } 

也可以通過以下方法，覆蓋默認線程池。

HConnection createConnection(org.apache.hadoop.conf.Configuration conf,ExecutorService pool); 

process query

Step1: 根據查詢條件，確定 RowKey 前綴

根據 3.3 RowKey 設計介紹，HBase 的寫和讀都遵循該設計規則。此處我們采用相同的方法，將 web 調用方傳入的查詢條件，轉化成對應的 RowKey 前綴。例如，查詢 check 表傳遞過來的 check_id=A208849559，生成前綴 7+7c9498b4a83974da56b252122b9752bf。

Step2：確定 scan 范圍

A208849559 對應的查詢結果數據即在 RowKey 前綴為 7+7c9498b4a83974da56b252122b9752bf 對應的 RowKey 及 value 中。

scan.setStartRow(Bytes.toBytes(rowkey_pre)); //scan, 7+7c9498b4a83974da56b252122b9752bf byte[] stopRow = Bytes.toBytes(rowkey_pre); stopRow[stopRow.length-1]++; scan.setStopRow(stopRow);// 7+7c9498b4a83974da56b252122b9752bg 

Step3：查詢結果組成返回對象

遍歷 ResultScanner 對象，將每一行對應的數據封裝成 table entity，組成 list 返回。

測試

從原始數據中隨機抓取 1000 個 check_id，用於模擬測試，連續發起 3 次請求數為 2000（200 個線程並發，循環 10 次），平均響應時間為 51ms，錯誤率為 0。

如上圖，經歷 N 次累計測試后，各個 region 上的 Requests 數較為接近，符合負載均衡設計之初。

踩坑記錄

1、kerberos 認證問題

如果集群開啟了安全認證，那么在進行 Spark 提交作業以及訪問 HBase 時，均需要進行 kerberos 認證。

本文采用 yarn cluster 模式，像提交普通作業一樣，可能會報以下錯誤。

ERROR StartApp: job failure,
java.lang.NullPointerException
	at com.tinawang.spark.hbase.utils.HbaseKerberos.<init>(HbaseKerberos.java:18)
	at com.tinawang.spark.hbase.job.SparkWriteHbaseJob.run(SparkWriteHbaseJob.java:60)

定位到 HbaseKerberos.java:18，代碼如下：

this.keytabPath = (Thread.currentThread().getContextClassLoader().getResource(prop.getProperty("hbase.keytab"))).getPath();

這是因為 executor 在進行 HBase 連接時，需要重新認證，通過 --keytab 上傳的 tina.keytab 並未被 HBase 認證程序塊獲取到，所以認證的 keytab 文件需要另外通過 --files 上傳。示意如下

--keytab /path/tina.keytab \
--principal [e-mail] \
--files "/path/tina.keytab.hbase"

其中 tina.keytab.hbase 是將 tina.keytab 復制並重命名而得。因為 Spark 不允許同一個文件重復上傳。

2、序列化

org.apache.spark.SparkException: Task not serializable
	at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:298)
	at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:288)
	at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:108)
	at org.apache.spark.SparkContext.clean(SparkContext.scala:2101)
	at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:370)
	at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:369)
	...
org.apache.spark.deploy.yarn.ApplicationMaster$$anon$2.run(ApplicationMaster.scala:637)
Caused by: java.io.NotSerializableException: org.apache.spark.api.java.JavaSparkContext
Serialization stack:
	- object not serializable (class: org.apache.spark.api.java.JavaSparkContext, value: org.apache.spark.api.java.JavaSparkContext@24a16d8c)
	- field (class: com.tinawang.spark.hbase.processor.SparkReadFileRDD, name: sc, type: class org.apache.spark.api.java.JavaSparkContext)
...

解決方法一：

如果 sc 作為類的成員變量，在方法中被引用，則加 transient 關鍵字，使其不被序列化。

private transient JavaSparkContext sc;

解決方法二：

將 sc 作為方法參數傳遞，同時使涉及 RDD 操作的類 implements Serializable。 代碼中采用第二種方法。詳見代碼。

3、批量請求測試

Exception in thread "http-nio-8091-Acceptor-0" java.lang.NoClassDefFoundError: org/apache/tomcat/util/ExceptionUtils

或者

Exception in thread "http-nio-8091-exec-34" java.lang.NoClassDefFoundError: ch/qos/logback/classic/spi/ThrowableProxy

查看下面 issue 以及一次排查問題的過程，可能是 open file 超過限制。

https://github.com/spring-projects/spring-boot/issues/1106

http://mp.weixin.qq.com/s/34GVlaYDOdY1OQ9eZs-iXg

使用 ulimit-a 查看每個用戶默認打開的文件數為 1024。

在系統文件 /etc/security/limits.conf 中修改這個數量限制，在文件中加入以下內容, 即可解決問題。

• soft nofile 65536
• hard nofile 65536

作者介紹

汪婷，中國民生銀行大數據開發工程師，專注於 Spark 大規模數據處理和 Hbase 系統設計。

參考文獻

http://hbase.apache.org/book.html#perf.writing
http://www.opencore.com/blog/2016/10/efficient-bulk-load-of-hbase-using-spark/
http://hbasefly.com/2016/03/23/hbase_writer/
https://github.com/spring-projects/spring-boot/issues/1106
http://mp.weixin.qq.com/s/34GVlaYDOdY1OQ9eZs-iXg