第三章 客戶(hù)端API: 基礎篇(第三部分)批量操作前面介紹了如何逐條和批量地添加、讀取、刪除數據。在這一節，我們將介紹如何一次執行多種不同類(lèi)型的操作處理多行記錄。
事實(shí)上，一些使用List結構的批量操作，如delete(List<Delete>
deletes)或者get(List(Get> gets)等，底層都是依靠batch實(shí)現的。封裝這些接口可以更好的提高友好性。
在下面介紹的批量操作中，您將會(huì )看到一個(gè)新的類(lèi)型叫Row，前面將到的Put、Get、Delete類(lèi)都是從Row類(lèi)的子類(lèi)。
void
batch(List<Row> actions, Object[] results)
throws
IOException, InterruptedException
Object[] batch(List<Row>
actions)
throws
IOException, InterruptedException
由于Row的存在，以及它和Get、Put、Delete的繼承關(guān)系，決定了可以在一個(gè)列表中混合多種不同類(lèi)型的操作。示例3-16給出了這種使用的例子。
值得注意地是，您不應該將一個(gè)row的Put和Delete操作混合在一起。因為L(cháng)ist中多個(gè)操作在服務(wù)器端執行的順序是無(wú)法保證的，這樣會(huì )得到一個(gè)無(wú)法預料到的結果。
示例3-16 批量操作
private final
static byte[] ROW1 = Bytes.toBytes(“row1″);
private final
static byte[] ROW2 = Bytes.toBytes(“row2″);
private final
static byte[] COLFAM1 = Bytes.toBytes(“colfam1″);
private final
static byte[] COLFAM2 = Bytes.toBytes(“colfam2″);
private final
static byte[] QUAL1 = Bytes.toBytes(“qual1″);
private final
static byte[] QUAL2 = Bytes.toBytes(“qual2″);
List<Row>
batch = new ArrayList<Row>();
Put put = new
Put(ROW2);
put.add(COLFAM2,
QUAL1, Bytes.toBytes(“val5″));
batch.add(put);
Get get1 = new
Get(ROW1);
get1.addColumn(COLFAM1,
QUAL1);
batch.add(get1);
Delete delete =
new Delete(ROW1);
delete.deleteColumns(COLFAM1,
QUAL2);
batch.add(delete);
Get get2 = new
Get(ROW2);
get2.addFamily(Bytes.toBytes(“BOGUS”));
batch.add(get2);
Object[] results =
new Object[batch.size()];
try {
table.batch(batch,
results);
} catch (Exception
e) {
System.err.println(“Error:
” + e);
}
for (int i = 0; i
< results.length; i++) {
System.out.println(“Result["
+ i + "]: ” + results[i]);
}
首先定義了一組指向row、column family、column
qualifier的常量，方便重用。然后創(chuàng )建一個(gè)Row實(shí)例的列表。向其中分別加入一個(gè)Put、Get、Delete實(shí)例，再添加一個(gè)指向不存在列的Put操作。創(chuàng )建一個(gè)Result數組，大小和batch的大小相同。然后執行batch方法，返回的結果放在result中，最后打印出Result數組的值。
整個(gè)程序的輸出如下：
Before batch
call…
KV:
row1/colfam1:qual1/1/Put/vlen=4, Value: val1
KV:
row1/colfam1:qual2/2/Put/vlen=4, Value: val2
KV: row1/colfam1:qual3/3/Put/vlen=4,
Value: val3
Result[0]:
keyvalues=NONE
Result[1]:
keyvalues={row1/colfam1:qual1/1/Put/vlen=4}
Result[2]:
keyvalues=NONE
Result[3]:
org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException:
org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException:
Column family
BOGUS does not exist in …
After batch
call…
KV:
row1/colfam1:qual1/1/Put/vlen=4, Value: val1
KV:
row1/colfam1:qual3/3/Put/vlen=4, Value: val3
KV:
row2/colfam2:qual1/1308836506340/Put/vlen=4, Value: val5
Error:
org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException:
Failed 1 action:
NoSuchColumnFamilyException: 1 time,
servers with
issues: 10.0.0.43:60020,
在前面的例子中，事先插入了一些測試數據并打印出來(lái)，讓您方便地看出示例代碼做的事情。最后發(fā)現，想刪除的記錄成功被刪除，想插入的新記錄也成功被插入了。
Get操作的結果，需要從Result數組中找到。Result數組的大小和請求操作的總數相等。第一個(gè)Result是Put操作的結果，為一個(gè)空的KeyValue結構;第二個(gè)Result的值是Get操作的結果，可以打印出它的值。第三個(gè)值為Delete操作的值，為一個(gè)空的KeyValue結構。表3-7給出了Result與Row類(lèi)型之前的對應關(guān)系。
表3-7 batch()調用可能返回的結果
Result
Description
null
The operation has failed to
communicate with the remote server.
Empty Result
Returned for successful Put
and Delete operations.
Result
Returned for successful Get
operations, but may also be empty when there was no matching row or column.
Throwable
In case the servers return an
exception for the operation it is returned to the client as-is. You can use
it to check what went wrong and maybe handle the problem automatically in
your code.
觀(guān)察示例3-16執行的結果，您可以發(fā)現空的Result對象打印出了keyvalues=NONE。 Get操作打印出了對應取到的值。對于錯誤列上的Put操作得到了一個(gè)異常。
值得注意地是，當您使用batch()方法時(shí)，里面的Put實(shí)例并不會(huì )使用客戶(hù)端緩存。batch()調用是同步的，同時(shí)直接向服務(wù)器發(fā)送請求。沒(méi)有時(shí)延和其它處理過(guò)程，這和批量put操作是不同的。因此，您需要選擇合適于您的。
batch調用有兩種形式，一種將Result數據放在參數中，另一種放在返回值中。
void
batch(List<Row> actions, Object[] results)
throws
IOException, InterruptedException
Object[]
batch(List<Row> actions)
throws
IOException, InterruptedException
它們之間比較大的區別在于，當拋出異常時(shí)，第一方法的result中被填充了部分結果。而第二個(gè)方法在異常時(shí)，將會(huì )返回null。
兩種方法都支持get、put、delete操作。如果執行其中的任何一個(gè)請求時(shí)出錯，一個(gè)客戶(hù)端異常將會(huì )被拋出，報告出錯情況?？蛻?hù)端緩存不會(huì )被使用到。void batch(actions,
results)會(huì )得到所有成功操作的結果和其中失敗的服務(wù)端異常。Object[] batch( actions )只返回客戶(hù)端異常，沒(méi)有結果被填充到返回值中。
所有的操作都將在check之前執行：當您發(fā)現一個(gè)action出現了錯誤，但其它的操作也將被執行。在最差的場(chǎng)景下，所有的action都返回失敗。
另一方面，batch操作在意瞬時(shí)失敗，比如NotServingRegionException（比如一個(gè)Region Server被下線(xiàn)了），它會(huì )進(jìn)行重試幾次。配置項hbase.client.retries.number會(huì )設定重試的次數（默認的重試次數為10）。
Row Locks(行鎖)
更新操作，比如put()、delete()、checkAndPut()等等對于一個(gè)row來(lái)說(shuō)是互斥執行的，從而保證一個(gè)低層面的原子性。Region Server提供了一個(gè)row lock行級鎖來(lái)保證只有擁有鎖的客戶(hù)端才能夠對該行進(jìn)行修改。在實(shí)際中，客戶(hù)端并不提供這種顯示的鎖，而是依賴(lài)于一種將每個(gè)操作獨立開(kāi)的機制。
您應該最大限度地避免使用row lock，很容易出現RDBMS中類(lèi)似的死鎖現象。當Lock在等待超時(shí)的過(guò)程中，兩個(gè)被掛起的客戶(hù)端都持有一個(gè)句柄，這類(lèi)句柄屬于稀缺資源。如果這個(gè)鎖被加在了一個(gè)訪(fǎng)問(wèn)很頻繁的行上，那么很多客戶(hù)端都會(huì )被阻塞。
當您使用put使用將一個(gè)Put實(shí)例發(fā)送到服務(wù)器時(shí)，如果您使用了如下的構造函數：
Put(byte[] row)
在這個(gè)構造函數里，并沒(méi)有出現RowLock參數，服務(wù)器會(huì )根據您的行為，自動(dòng)為您創(chuàng )建一個(gè)鎖。這個(gè)鎖對于客戶(hù)端是透明的，客戶(hù)端無(wú)法獲取到這個(gè)生命期很短的服務(wù)器端的鎖。相比于服務(wù)器自動(dòng)創(chuàng )建的隱式鎖?？蛻?hù)端也可以使用顯式鎖，并且在多個(gè)操作中使用到它?？梢酝ㄟ^(guò)以下的方法：
RowLock
lockRow(byte[] row) throws IOException
void
unlockRow(RowLock rl) throws IOException
第一個(gè)方法lockRow以rowkey為參數，返回一個(gè)RockLock實(shí)例，這個(gè)實(shí)例可以傳遞到Put和Delete的構造函數中去。一旦您成功的獲取到一個(gè)鎖后，在使用完之后，您必須調用unlockRow方法釋放它。
您申請到的LockRow會(huì )鎖定整行，它通過(guò)rowkey來(lái)確定行，一旦擁有，別的客戶(hù)端將不能對該行進(jìn)行更新操作。
當這個(gè)行鎖無(wú)論是被客戶(hù)端或者服務(wù)器占有時(shí)，另一個(gè)想申請一個(gè)該行的新行鎖的行為都會(huì )被掛起，直到這行原有的行鎖被釋放或者過(guò)期。后者是對出錯進(jìn)程持有行鎖情況的一個(gè)兼容。
一個(gè)行鎖的默認超時(shí)時(shí)間是1分鐘，可以在hbase-site.xml進(jìn)行配置：
<property>
<name>hbase.regionserver.lease.period</name>
<value>120000</value>
</property>
上述代碼將超時(shí)時(shí)間改為120秒。當然，這個(gè)值不能設的太大，否則，每個(gè)客戶(hù)端申請一個(gè)被別的進(jìn)程占用的鎖的最大等待時(shí)間都會(huì )變?yōu)檫@個(gè)值。示例3-17給出一個(gè)由用戶(hù)生成的行鎖阻塞其它讀取客戶(hù)端的例子。
示例3-17 使用顯示行鎖
static class
UnlockedPut implements Runnable {
@Override
public void run()
{
try {
HTable
table = new HTable(conf, “testtable”);
Put
put = new Put(ROW1);
put.add(COLFAM1,
QUAL1, VAL3);
long
time = System.currentTimeMillis();
System.out.println(“Thread
trying to put same row now…”);
table.put(put);
System.out.println(“Wait
time: ” +
(System.currentTimeMillis()
- time) + “ms”);
} catch
(IOException e) {
System.err.println(“Thread
error: ” + e);
}
}
}
System.out.println(“Taking
out lock…”);
RowLock lock =
table.lockRow(ROW1);
System.out.println(“Lock
ID: ” + lock.getLockId());
Thread thread =
new Thread(new UnlockedPut());
thread.start();
try {
System.out.println(“Sleeping
5secs in main()…”);
Thread.sleep(5000);
} catch
(InterruptedException e) {
//
ignore
}
try {
Put put1 = new
Put(ROW1, lock);
put1.add(COLFAM1,
QUAL1, VAL1);
table.put(put1);
Put put2 = new
Put(ROW1, lock);
put2.add(COLFAM1, QUAL1,
VAL2);
table.put(put2);
} catch (Exception
e) {
System.err.println(“Error:
” + e);
} finally {
System.out.println(“Releasing
lock…”);
table.unlockRow(lock);
}
首先定義了一個(gè)的線(xiàn)程，它會(huì )不斷使用隱式鎖不斷地訪(fǎng)問(wèn)同一行上的記錄，在取得鎖之前，put操作始終會(huì )被掛起，在線(xiàn)程內部會(huì )打印出等待的時(shí)間。
主線(xiàn)程，先顯示的創(chuàng )建一個(gè)行鎖，然后啟動(dòng)一個(gè)前面定義的線(xiàn)程實(shí)例。接著(zhù)，主線(xiàn)程sleep 5秒之后，將這個(gè)鎖傳給Put對象，進(jìn)行put操作之后，這個(gè)鎖會(huì )被釋放，從而前面那個(gè)無(wú)鎖的線(xiàn)程會(huì )繼續運行。
運行這段示例代碼，將會(huì )得到如下的輸出：
Taking out lock…
Lock ID:
4751274798057238718
Sleeping 5secs in
main()…
Thread trying to
put same row now…
Releasing lock…
Wait time: 5007ms
After thread
ended…
KV:
row1/colfam1:qual1/1300775520118/Put/vlen=4, Value: val2
KV:
row1/colfam1:qual1/1300775520113/Put/vlen=4, Value: val1
KV:
row1/colfam1:qual1/1300775515116/Put/vlen=4, Value: val3
可以看出無(wú)鎖的線(xiàn)程的確被阻塞了5秒鐘。直到主線(xiàn)程做完兩次put操作后釋放了行鎖，才繼續運行。我們可以看到一個(gè)有趣的現象，由無(wú)鎖對象最后插入的值，卻擁有最小的時(shí)間戳。這是因為，實(shí)際上無(wú)鎖線(xiàn)程的put操作是最早執行的。一旦它被發(fā)送到Region Server服務(wù)器，Region Server便會(huì )給它打上一個(gè)時(shí)間戳，雖然此時(shí)它還無(wú)法獲取到行鎖而被阻塞，但此時(shí)的時(shí)間戳已經(jīng)生成了。主線(xiàn)程一共花了7ms向服務(wù)器提交了兩次Put命令，并釋放了行鎖。
當您使用一個(gè)先前申請到的行鎖，但是由于超時(shí)無(wú)效時(shí)，您將會(huì )收到服務(wù)器端生成的一個(gè)錯誤，以UnknownRowLockException異常的形式返回。它說(shuō)明，服務(wù)器端已經(jīng)將這個(gè)鎖廢棄掉了。此時(shí)，您可以將它drop掉，然后重新申請一個(gè)新的鎖。
Scan(掃描)前面我們討論了基本的CRUD類(lèi)型的操作，現在輪到scan了，類(lèi)似于數據庫系統中的游標(cursor)。它可以充分使用到HBase提供的順序性的、排序的存儲結構。
基本介紹
scan操作與get操作非常相似。介紹它之前，必須先介紹另一個(gè)類(lèi)Scan。由于scan更像是一個(gè)迭代器，因此，并沒(méi)有scan()的方法，取而代之的是getScanner方法，它會(huì )返回您想要遍歷訪(fǎng)問(wèn)的scanner對象?？梢岳萌缦碌姆椒ㄈ〉剿?div style="height:15px;">
ResultScanner
getScanner(Scan scan) throws IOException
ResultScanner
getScanner(byte[] family) throws IOException
ResultScanner
getScanner(byte[] family, byte[] qualifier)
throws
IOException
后面兩個(gè)方法是為了加強友好性，會(huì )隱式地創(chuàng )建一個(gè)scan對象，然后再調用getScanner(Scan
scan)方法取到ResultScanner。
Scan類(lèi)有如下的構造函數：
Scan()
Scan(byte[]
startRow, Filter filter)
Scan(byte[]
startRow)
Scan(byte[]
startRow, byte[] stopRow)
與Get方法不同的是，您不再指定一個(gè)具體的rowkey，您現在可以選擇性的指定一個(gè)startRow參數，這個(gè)參數定義HTable中，要讀取的rowkey的起始位置?？蛇x的參數stopRow定義讀取rowkey的結束位置。
startRow是被包含的，而endRow是不被包含的，因此，可以用以表述式[startRow,
stopRow)來(lái)描述這種關(guān)系。
Scan提供的一個(gè)特殊的功能就是您不必精確地指定rowkey。相反，scan會(huì )相匹配與指定的startKey相等或者大于它的所有的rowkey。如是沒(méi)有指定startKey，那么將會(huì )從表的起始位置開(kāi)始遍歷。
如果指定了stopKey，那么只會(huì )遍歷rowkey小于stopKey的所有記錄。如果沒(méi)有指定stopKey，那么scan會(huì )遍歷到表的末尾。
還有另一個(gè)可選的參數Filter，這個(gè)參數指向一個(gè)Filter實(shí)例。Scan對象常常使用空的構造函數來(lái)創(chuàng )建，另外的參數都可以通過(guò)getter和setter方法進(jìn)行指定。
Scan實(shí)例被創(chuàng )建后，如果想加入更多的限制條件，您可以使用很多的方法來(lái)限制讀出的數據。當然，您也可以創(chuàng )建一個(gè)空的Scan，將整個(gè)表的所有的column family和column
qualifier讀出來(lái)。
Scan
addFamily(byte [] family)
Scan
addColumn(byte[] family, byte[] qualifier)
和Get類(lèi)相似，您也可以使用addFamily來(lái)設置column
families或者使用addColumn來(lái)設置column，從而限定讀出數據的條件。
如果您只需要數據的一部分，那么通過(guò)限制Scan的范圍，只是體現了HBase的強大之處，由于數據是以column family為物理單元分隔文件的，不在Scan范圍內的column family對應的文件根本不會(huì )被打開(kāi)，這只是面向列存儲最大的優(yōu)勢所在。
Scan
setTimeRange(long minStamp, long maxStamp) throws IOException
Scan
setTimeStamp(long timestamp)
Scan
setMaxVersions()
Scan setMaxVersions(int
maxVersions)
您還可以限制Scan的timestamp，設置timestamp的范圍，設置掃描的版本數。使用setStartRow()，setStopRow()和setFilter()，可以達到構造函數中的參數值同樣的效果。方法hasFilter()可以判斷一個(gè)Scan中是否添加了filter。表3-8列出了其它一些方法。
表3-8 Scan的方法一覽
Result
Description
getStartRow()/getStopRow()
Can be used to retrieve the
currently assigned values.
getTimeRange()
Retrieves the associated timestamp or time range of the Get
instance. Note that there is no getTimeStamp() since the API converts a value
assigned with setTimeStamp() into a TimeRange instance internally, setting
the minimum and maximum values to the given timestamp.
getMaxVersions()
Returns the currently configured number of versions that should
be retrieved from the table for every column.
getFilter()
Special filter instances can be used to select certain columns
or cells, based on a wide variety of conditions. You can get the currently
assigned filter using this method. It may return null if none was previously
set.
setCacheBlocks()
/getCacheBlocks()
Each HBase region server has a block cache that efficiently
retains recently accessed data for subsequent reads of contiguous
information. In some events it is better to not engage the cache to avoid too
much churn when doing full table scans. These methods give you control over
this feature.
numFamilies()
Convenience method to retrieve the size of the family map,
containing the families added using the addFamily() or addColumn() calls.
hasFamilies()
Another helper to check if a family—or column—has been added to the current
instance of the Scan class.
getFamilies()
/setFamilyMap()
/getFamilyMap()
These methods give you access to the column families and
specific columns, as added by the addFamily() and/or addColumn() calls. The
family map is a map where the key is the family name and the value is a list
of added column qualifiers for this particular family. The getFamilies()
returns an array of all stored families, i.e., containing only the family
names (as
byte[] arrays).
當您創(chuàng )建了Scan實(shí)例之后，您就需要調用HTable的getScanner方法，取得ResultScanner實(shí)例。在下一節中，我們將詳細介紹ResultScanner類(lèi)。
ResultScanner類(lèi)Scans不會(huì )將所有匹配到的行通過(guò)一個(gè)RPC調用發(fā)送到客戶(hù)端，而是分多次。這是因為，一行數據可能很大，放在一次傳輸會(huì )消耗掉很多資源、花費很長(cháng)的時(shí)間。
ResultScanner將scan轉化成一種類(lèi)似于get的操作，將結果通過(guò)迭代器訪(fǎng)問(wèn)出來(lái)。它具有自己的一些方法：
Result next()
throws IOException
Result[] next(int
nbRows) throws IOException
void close()
有兩種形式的next函數可以調用，close()操作用來(lái)釋放資源。
Next()調用返回一個(gè)單獨的Result實(shí)例，存放一個(gè)可用的row對象。同樣的，您也可以使用next(int
nbRows)一次取回來(lái)很多行，該調用返回一個(gè)Result的數組對象，數組中的每一行都代表一個(gè)唯一的row。當然，取到的值可能小于nbRows，但這很少在未取完數據時(shí)發(fā)生?？梢圆榭?，前面對Result實(shí)例的介紹，學(xué)習如何使用這個(gè)類(lèi)。示例3-18給出了如何使用scan訪(fǎng)問(wèn)一個(gè)表。
示例3-18 使用scan訪(fǎng)問(wèn)數據
Scan scan1 = new
Scan();
ResultScanner
scanner1 = table.getScanner(scan1);
for (Result res :
scanner1) {
System.out.println(res);
}
scanner1.close();
Scan scan2 = new
Scan();
scan2.addFamily(Bytes.toBytes(“colfam1″));
ResultScanner
scanner2 = table.getScanner(scan2);
for (Result res :
scanner2) {
System.out.println(res);
}
scanner2.close();
Scan scan3 = new
Scan();
scan3.addColumn(Bytes.toBytes(“colfam1″),
Bytes.toBytes(“col-5″)).
addColumn(Bytes.toBytes(“colfam2″),
Bytes.toBytes(“col-33″)).
setStartRow(Bytes.toBytes(“row-10″)).
setStopRow(Bytes.toBytes(“row-20″));
ResultScanner
scanner3 = table.getScanner(scan3);
for (Result res :
scanner3) {
System.out.println(res);
}
scanner3.close();
首先創(chuàng )建一個(gè)空的Scan對象，使用這個(gè)對象對表進(jìn)行遍歷，然后關(guān)閉這個(gè)scanner釋放相關(guān)資源。接著(zhù)再創(chuàng )建一個(gè)只查詢(xún)colfam1下記錄的scanner，并打印相關(guān)的記錄。最后創(chuàng )建一個(gè)只掃描列colfam1:col-5和colfam2:col-33，且rowkey范圍從row-10到row-20的所有記錄。
為了測試上述示例，首先創(chuàng )建一個(gè)表，含有colfam1和colfam2兩個(gè)column family。然后很這個(gè)表中插入100行記錄。我們不列出全表掃描的輸出結果，而僅列出scan2和scan3的輸出結果：
Scanning table
#3…
keyvalues={row-10/colfam1:col-5/1300803775078/Put/vlen=8,
row-10/colfam2:col-33/1300803775099/Put/vlen=9}
keyvalues={row-100/colfam1:col-5/1300803780079/Put/vlen=9,
row-100/colfam2:col-33/1300803780095/Put/vlen=10}
keyvalues={row-11/colfam1:col-5/1300803775152/Put/vlen=8,
row-11/colfam2:col-33/1300803775170/Put/vlen=9}
keyvalues={row-12/colfam1:col-5/1300803775212/Put/vlen=8,
row-12/colfam2:col-33/1300803775246/Put/vlen=9}
keyvalues={row-13/colfam1:col-5/1300803775345/Put/vlen=8,
row-13/colfam2:col-33/1300803775376/Put/vlen=9}
keyvalues={row-14/colfam1:col-5/1300803775479/Put/vlen=8,
row-14/colfam2:col-33/1300803775498/Put/vlen=9}
keyvalues={row-15/colfam1:col-5/1300803775554/Put/vlen=8,
row-15/colfam2:col-33/1300803775582/Put/vlen=9}
keyvalues={row-16/colfam1:col-5/1300803775665/Put/vlen=8,
row-16/colfam2:col-33/1300803775687/Put/vlen=9}
keyvalues={row-17/colfam1:col-5/1300803775734/Put/vlen=8,
row-17/colfam2:col-33/1300803775748/Put/vlen=9}
keyvalues={row-18/colfam1:col-5/1300803775791/Put/vlen=8,
row-18/colfam2:col-33/1300803775805/Put/vlen=9}
keyvalues={row-19/colfam1:col-5/1300803775843/Put/vlen=8,
row-19/colfam2:col-33/1300803775859/Put/vlen=9}
keyvalues={row-2/colfam1:col-5/1300803774463/Put/vlen=7,
row-2/colfam2:col-33/1300803774485/Put/vlen=8}
再一次強調的時(shí)，輸出的結果與插入的順序無(wú)關(guān)，有趣的是，rowkey的排列按照了字母序進(jìn)行輸出。
Caching vs. Batching以目前為止，每個(gè)next()操作都是一次RPC調用，即使當你使用next( int
nbRows)時(shí)。因為這個(gè)next( int nbRows )只是一個(gè)簡(jiǎn)單的next()調用的客戶(hù)端循環(huán)。顯然，這樣的性能是很難令人滿(mǎn)意的。因此，在一次RPC調用中返回多行便顯得意義重大。這個(gè)機制叫制scanner緩存(scanner
caching)，它默認是關(guān)閉的。
您可以使用在兩個(gè)層面打開(kāi)它：在表一級設定，將會(huì )對這個(gè)表的所有的scanner實(shí)例生效。在scan一級設定，將只會(huì )對這個(gè)scan生效。當然您也可以對通過(guò)對HTable實(shí)例進(jìn)行設定，達到對所有的表生效。
void
setScannerCaching(int scannerCaching)
int
getScannerCaching()
當然您也可以在HBase安裝時(shí)，修改默認值?？梢酝ㄟ^(guò)修改配置文件hbase-site.xml實(shí)現：
<property>
<name>hbase.client.scanner.caching</name>
<value>10</value>
</property>
上述配置文件將cache的默認大小改為了10。當然，您還可以在代碼中繼續設定新的值。
在您通過(guò)getScanner方法，得到一個(gè)Scanner對象后，通過(guò)setScannerCaching()設置緩存大小，getScannerCaching()得到目前的緩存大小。API將會(huì )把設置的大小傳遞給scanner對象，除非您使用了Scan類(lèi)的方法：
void
setCaching(int caching)
int getCaching()
scan直接設置cache大小擁有最高的優(yōu)先級。通過(guò)對緩存大小的設定，可以使一次RPC調用返回多行記錄。兩種next()都會(huì )使用到這個(gè)緩存。
您可能需要找到RPC操作的數據和內存占用情況的一個(gè)折中，scanner的緩存大小越大，讀取的性能越好(當然值過(guò)大，也不好)，但緩存的條目多了之后，一次傳輸消耗的時(shí)間越長(cháng)，占用的堆空間大小也越大，還會(huì )引發(fā)OOM異常(OutOfMemoryException)。
當從服務(wù)器到客戶(hù)端傳輸數據的時(shí)間或者客戶(hù)端處理數據的時(shí)間大于了scanner設置的超時(shí)時(shí)間，那么客戶(hù)端將會(huì )收到一個(gè)ScannerTimeoutException。
示例3-19 scanner超時(shí)的例子
Scan scan = new
Scan();
ResultScanner
scanner = table.getScanner(scan);
int scannerTimeout
= (int) conf.getLong(
HConstants.HBASE_REGIONSERVER_LEASE_PERIOD_KEY,
-1);
try {
Thread.sleep(scannerTimeout
+ 5000);
} catch
(InterruptedException e) {
//
ignore
}
while (true){
try {
Result
result = scanner.next();
if
(result == null) break;
System.out.println(result);
} catch (Exception
e) {
e.printStackTrace();
break;
}
}
scanner.close();
首先獲取當前scanner的超時(shí)時(shí)間，然后sleep一會(huì )兒，等待超時(shí)。接著(zhù)嘗試打印出取到的結果集。將會(huì )得到如下的輸出：
Adding rows to
table…
Current (local)
lease period: 60000
Sleeping now for
65000ms…
Attempting to
iterate over scanner…
Exception in
thread “main” java.lang.RuntimeException:
org.apache.hadoop.hbase.client.ScannerTimeoutException:
65094ms passed
since
the last invocation, timeout is currently set to 60000
at
org.apache.hadoop.hbase.client.HTable$ClientScanner$1.hasNext
at
ScanTimeoutExample.main
Caused by:
org.apache.hadoop.hbase.client.ScannerTimeoutException: 65094ms
passed
since the last invocation, timeout is currently set to 60000
at
org.apache.hadoop.hbase.client.HTable$ClientScanner.next
at
org.apache.hadoop.hbase.client.HTable$ClientScanner$1.hasNext
…
1 more
Caused by:
org.apache.hadoop.hbase.UnknownScannerException:
org.apache.hadoop.hbase.UnknownScannerException:
Name: -315058406354472427
at
org.apache.hadoop.hbase.regionserver.HRegionServer.next
…
在scanner超時(shí)之后，客戶(hù)端嘗試打印從服務(wù)器上取出的值時(shí)，將會(huì )將到一個(gè)異常。
您可能想在客戶(hù)端代碼中加入以下代碼來(lái)增大超時(shí)時(shí)間：
Configuration conf
= HBaseConfiguration.create()
conf.setLong(HConstants.HBASE_REGIONSERVER_LEASE_PERIOD_KEY,
120000)
但由于scan的超時(shí)時(shí)間是配在Region Server上的，因此，上述配置并不會(huì )生效。如果您真的想修改這個(gè)值，您只有去Region Server上去修改hbase-site.xml，并重啟集群。
從輸出打印的堆?？梢钥闯?，ScannerTimeoutException嵌套在了UnknownScannerException之中，這意味著(zhù)next()調用使用了一個(gè)已經(jīng)過(guò)期的scanner ID，但這個(gè)ID已經(jīng)被刪除了。換句話(huà)說(shuō)，客戶(hù)端存儲的scanner ID，Region Server已經(jīng)不認識了，從而拋出一個(gè)UnknownScannerException。
現在，您已經(jīng)學(xué)會(huì )了如何使用客戶(hù)端的scanner緩存來(lái)提高批量交互的性能。但有一點(diǎn)要注意的是，對于非常大的行，可能無(wú)法放入客戶(hù)端的內存中。使用HBase客戶(hù)端API中的batching，可以處理這種情況：
void setBatch(int
batch)
int getBatch()
與caching(處理層次為rows)相對應，batching處理的層次是columns。它控制一次next()調用傳輸多少個(gè)columns。通過(guò)ResultScanner的setBatch()方法可以進(jìn)行設置，setBatch(5)將使每個(gè)Result實(shí)例，返回5個(gè)columns。
當一行含有非常多的列時(shí)，您可以使用setBatch方法，一次next()返回一行中的部分列。Result中返回的列也可能達不到batching的值，比較一行有17列，batching的值為5，那么前三次next()將得到5,5,5，最后一次調用只能返回2個(gè)列。
通過(guò)設置caching和batch的大小，scanner可以在選擇rowkey范圍查詢(xún)時(shí)控制RPC的多少。示例3-20用兩個(gè)參數來(lái)對Result實(shí)例大小與請求次數進(jìn)行調優(yōu)。
示例3-20 使用caching和batch兩個(gè)參數
private static
void scan(int caching, int batch) throws IOException {
Logger log =
Logger.getLogger(“org.apache.hadoop”);
final int[]
counters = {0, 0};
Appender appender
= new AppenderSkeleton() {
@Override
protected void
append(LoggingEvent event) {
String msg =
event.getMessage().toString();
if (msg != null
&& msg.contains(“Call: next”)) {
counters[0]++;
}
}
@Override
public void
close() {}
@Override
public boolean
requiresLayout() {
return
false;
}
};
log.removeAllAppenders();
log.setAdditivity(false);
log.addAppender(appender);
log.setLevel(Level.DEBUG);
Scan scan = new
Scan();
scan.setCaching(caching);
scan.setBatch(batch);
ResultScanner
scanner = table.getScanner(scan);
for (Result result
: scanner) {
counters[1]++;
}
scanner.close();
System.out.println(“Caching:
” + caching + “, Batch: ” + batch +
“,
Results: ” + counters[1] + “, RPCs: ” + counters[0]);
}
public static void
main(String[] args) throws IOException {
scan(1, 1);
scan(200, 1);
scan(2000, 100);
scan(2, 100);
scan(2, 10);
scan(5, 100);
scan(5, 20);
scan(10, 10);
}
示例代碼首先設置caching和batch的參數，然后打印Result的大小和RPC的次數。對不同的caching和batch大小進(jìn)行了組合測試。
Caching: 1, Batch:
1, Results: 200, RPCs: 201
Caching: 200,
Batch: 1, Results: 200, RPCs: 2
Caching: 2000,
Batch: 100, Results: 10, RPCs: 1
Caching: 2, Batch:
100, Results: 10, RPCs: 6
Caching: 2, Batch:
10, Results: 20, RPCs: 11
Caching: 5, Batch:
100, Results: 10, RPCs: 3
Caching: 5, Batch:
20, Results: 10, RPCs: 3
Caching: 10,
Batch: 10, Results: 20, RPCs: 3
通過(guò)調整兩個(gè)參數的值，可以觀(guān)察它們對結果的影響。表3-9給出了一些組合的結果。為了運行示例3-20，首先創(chuàng )建了一個(gè)擁有兩個(gè)column family的表，添加了10行，每行中，每個(gè)column family下添加10個(gè)column。這就意味著(zhù)一共存在著(zhù)200個(gè)columns或者叫做cell，每個(gè)cell只有一個(gè)版本。
表3-9 示例參數的影響
Caching
Batch
Results
RPCs
Description
1
1
200
201
Each column is returned as a
separate Result instance. One more RPC is needed to realize the scan is
complete.
200
1
200
2
Each column is a separate
Result, but they are all transferred in one RPC (plus the extra check).
2
10
20
11
The batch is half the row
width, so 200 divided by 10 is 20 Results needed. 10 RPCs (plus the check) to
transfer them.
5
100
10
3
The batch is too large for
each row, so all 20 columns are batched. This requires 10 Result instances.
Caching brings the number of RPCs down to two (plus the check).
5
20
10
3
This is the same as above, but
this time the batch matches the columns available. The outcome is the same.
10
10
20
3
This divides the table into
smaller Result instances, but larger caching also means only two RPCs are
needed.
為了計算RPC的次數，您需要首先將行數與最行的column數相乘，然后用這個(gè)值除以batch和column數中的較小值。最后用這個(gè)值除以caching大小。用數學(xué)公式表示為：
RPCs = (Rows *
Cols per Row) / Min(Cols per Row, Batch Size) /Scanner Caching
還需要額外的RPC操作來(lái)打開(kāi)和關(guān)閉scanner。因此，還scanner還需要兩次額外的RPC操作。
圖3-2描述了caching和batching是如何起作用的。如圖所示，該表具有9行值，每行都有不定數目的column。設置scanner中caching的大小為6，batch大小為3。您可以看到，需要3次RPC操作來(lái)轉輸數據（虛線(xiàn)包圍的部分）。
圖3-2 通過(guò)caching和batching控制scan操作RPC的次數
由于batch大小小于一行中column的數目，因此，服務(wù)器將3個(gè)columns打成一個(gè)Result，一次RPC操作可以傳輸6個(gè)這樣的Result。如果batch大小不設置，而caching大小被設置時(shí)，每行記錄將包含一行中所有column，這樣一個(gè)Result實(shí)例中就是一個(gè)完整的行。只有當您設置了batch參數，才有可能把一行拆成多個(gè)Result實(shí)例。
一開(kāi)始您可能不需要考慮caching和batching的大小設置，當您進(jìn)行應用程序的調優(yōu)時(shí)，您必須對這個(gè)原理非常清楚才能找到一個(gè)最好的平衡點(diǎn)。
輔助功能
在進(jìn)一步了解客戶(hù)端的其它功能之前，我們有必要先了解HBase和它的客戶(hù)端提供的一些有用的輔助功能。
HTable方法集
客戶(hù)端API代表了一個(gè)HTable實(shí)例，它提供了訪(fǎng)問(wèn)一個(gè)HBase表的一些方法。除了前面提到的一些對于訪(fǎng)問(wèn)HBase表的主要方法，還有另外一些值得留意的方法：
void close()
該方法前面有所提及，但考慮到它的重要性，這樣有必要專(zhuān)門(mén)再次提及。在結束了對表的訪(fǎng)問(wèn)之后，一定要調用close()接口。當close()被調用時(shí)，客戶(hù)端會(huì )調用flushCache()方法，將客戶(hù)端緩存區中緩存的數據提交到服務(wù)器。
byte[]
getTableName()
這個(gè)方法可以方例地取出表名。
HTableDescriptor
getTableDescriptor()
HBase中每個(gè)表都會(huì )使用一個(gè)HTableDescriptor的實(shí)例。您可以通過(guò)getTableDescriptor()獲取對表信息的訪(fǎng)問(wèn)。
static Boolean
isTableEnabled(table)
HTable有4個(gè)靜態(tài)方法，它們都需要一個(gè)配置對象，如沒(méi)有提供configuration，HTable會(huì )在程序的classpath下使用一個(gè)默認的configuration。該函數檢查ZooKeeper上table表是否為enable狀態(tài)。
byte[][]
getStartKeys()
byte[][]
getEndKeys()
pair<byte[][],
byte[][]> getStartEndKeys()
這幾個(gè)函數可以訪(fǎng)問(wèn)表中當前的rowkey范圍，隨著(zhù)數據的不斷增加，調用后也會(huì )得到不同的結果。這3個(gè)方法返回byte數組。您可以使用Bytes.toStringBinary()來(lái)打印出key值。
void
clearRegionCache()
HRegionLocation
getRegionLocation( row )
Map<HRegionInfo,
HServerAddress> getRegionInfo()
這幾個(gè)方法使您可以取出Region的信息，您可以使用第一個(gè)方法來(lái)清楚客戶(hù)端上的緩存，也可以使用第三個(gè)方法來(lái)取出所有Region信息。這些方法幫助一些高級使用者來(lái)利用Region信息，比較路由負載、計算靠近數據等。
void
prewarmRegionCache(Map<HRegionInfo, HServerAddress> regionMap)
static void
setRegionCachePrefetch(table, enable)
static Boolean
getRegionCachePrefetch(table)
這也是一組高級用法的API。這組API可以提前將table的Region信息緩存到客戶(hù)端。使用上述API，您可以提供一個(gè)Region的列表來(lái)對Region信息進(jìn)行預熱。
Bytes類(lèi)
該類(lèi)用來(lái)將Java類(lèi)型，比如String或者long轉化為raw、byte數組等HBase支持的類(lèi)型。因此，再介紹這個(gè)類(lèi)和它的函數是有意義的。
大多數方法都有這三種形式，比如：
static long
toLong(byte[] bytes)
static long
toLong(byte[] bytes, int offset)
static long
toLong(byte[] bytes, int offset, int length)
它們的輸出都是byte數組，偏移量、長(cháng)度，后兩個(gè)可以缺省。它們的使用方式取決于您擁有的byte數組。如果您是使用Bytes.toBytes()方法得到的，那么您可以安全的使用第一個(gè)API，整個(gè)bytes數據存放著(zhù)待轉化的值。
在HBase內部，將數據存放在一個(gè)大的字節數組中，使用如下的方法：
static int
putLong(byte[] bytes, int offset, long val)
這個(gè)方法允許您將一個(gè)Long對象寫(xiě)入到一個(gè)給定的字節數組中的指定位置。如果您想從一個(gè)大數組中取出數據，可以使用toLong方法。
Bytes類(lèi)支持的Java類(lèi)型包括：String、Boolean、short、int、long、double、float。除了這些，表3-10中還列出了一些有用的方法。
表3-10 Bytes提供的一些方法
Result
Description
toStringBinary()
While working very similar to
toString(), this variant has an extra safeguard to convert nonprintable data
into their human-readable hexadecimal numbers. Whenever you are not sure what
a byte array contains you should use this method to print its content, for
example, to the console, or into a logfile.
compareTo()/equals()
These methods allow you to
compare two byte[], that is, byte arrays. The former gives you a comparison
result and the latter a boolean value, indicating whether the given arrays
are equal to each other.
add()/head()/tail()
You can use these to add two
byte arrays to each other, resulting in a new, concatenated array, or to get
the first, or last, few bytes of the given byte array.
binarySearch()
This performs a binary search
in the given array of values. It operates on byte arrays for the values and
the key you are searching for.
incrementBytes()
This increments a long value
in its byte array representation, as if you had used toBytes(long) to create
it. You can decrement using a negative amount parameter.
Bytes提供的一些方法與Java提供的ByteBuffer有一些復疊。區別是前者在處理的過(guò)程中不會(huì )生成新的實(shí)例，因此，它采用了一些優(yōu)化。對于HBase中，這種類(lèi)型與字節之間的轉化操作被頻繁使用，因此，通過(guò)這種優(yōu)化，可以避免非常耗時(shí)的GC操作。
