HTable에 write하기 위해서는 Put 객체를 생성하고 put 함수를 호출하면된다.
하나 row를 Put 하기
void put(Put put) throws IOException
Put(byte[] row)
Put(byte[] row, RowLock rowLock)
Put(byte[] row, long ts)
Put(byte[] row, long ts, RowLock rowLock)
예)
Configuration conf = HBaseConfiguration.create();
HTable table = new HTable(conf, "testtable");
Put put = new Put (Bytes.toBytes("row1"));
put.add((Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"), Bytes.toBytes("val1");
table.put(put);
Get,에서도 설명했듯이 byte 배열형태로 저장된다. 형변환을 쉽게 해주기위한 Bytes라는 클래스를 제공해준다.
static byte[] toBytes(ByteBuffer bb)
static byte[] toBytes(String s)
static byte[] toBytes(boolean b)
static byte[] toBytes(long val)
static byte[] toBytes(float f)
static byte[] toBytes(int val)
...
Put 객체에 특정 셀이 존재하는지 확인하기 위한 함수도 제공한다. Put.get, Put.getFamilyMap 메소드를 호출해서 확인하지 않고 별도의 함수를 제공한다. (HTable.get 메소드 또는 HTable.exists 함수와는 다른 것임) - 잘 쓰이지는 않는 함수들
boolean has(byte[] family, byte[] qualifier)
boolean has(byte[] family, byte[] qualifier, long ts)
boolean has(byte[] family, byte[] qualifier, byte[] value)
boolean has(byte[] family, byte[] qualifier, long ts, byte[] value)
Put 실행가능한 예제
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
public class PutExample {
public static void main(String[] args) throws IOException {
Configuration conf = HBaseConfiguration.create();
HTable table = new HTable(conf, "testtable");
Put put = new Put(Bytes.toBytes("row1"));
put.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"),
Bytes.toBytes("val1"));
put.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual2"),
Bytes.toBytes("val2"));
table.put(put);
}
}
Configuration conf = HBaseConfiguration.create();
설정을 생성한다. HBase 0.20.xx ---> HBase 0.90.xx 버전업되면서 API가 살짝 바뀌었다.
예전에는 new HBaseConfiguration() 했지만 이제 HBaseConfiguration.create() 호출
Configuration 을 생성하기 위하여 다음 두가지 API를 제공한다.
static Configuration create() - 자바클래스패스 안에 hbase-default.xml 과 hbase-site.xml 파일을 읽어서 속성값들을 지정한다.
static Configuration create(Configuration that) - 인수로 지정한 Configuration을 가장 우선순위가 높아서 기존 속성(파일들로 부터 읽은 속성)들을 재설정하게 된다.
Hadoop Job에 Configuration을 지정할 때도 HBaseConfiguration을 이용할 수 있다. (Configuration을 상속해서 만들었음)
Configuration 인스턴스는 쓰레드당 1개를 사용하도록 하는것이 바람직하다.
Connection Pool 섹션에서 설명되어 있다.
제대로 write가 되었는지 확인하기위해 명령행 셀을 이용할 수 있다.
$HBASE_HOME/bin/hbase shell
hbase(main):001:0> list
TABLE
testtable
1 row(s) in 0.0400 seconds
hbase(main):002:0> scan 'testtable'
ROW COLUMN+CELL
row1 column=colfam1:qual1, timestamp=1294065304642, value=val1
1 row(s) in 0.2050 seconds
list 명령으로 테이블들을 조회할 수 있다.
scan 명령으로 테이블에 저장된 값들을 확인할 수 있다.
HBase가 특징중에 하나로 데이터의 버전을 관리한다.
Timestamp를 가지고 이를 구분한다.
put 오퍼레이션이 수행될 때 RegionServer에 의해 Timestamp값이 채워진다. 그러므로 서버들은 적절한 시간을 가지고 있고 서로 동기화되어야만한다. (클라이언트가 지정할 수도 있지만 비추)
그럼 어떻게 여러 버전을 저장하고 조회 할 수 있는지 확인하자.
hbase(main):001:0> create 'test', 'cf1'
0 row(s) in 0.9810 seconds
hbase(main):002:0> put 'test', 'row1', 'cf1', 'val1'
0 row(s) in 0.0720 seconds
hbase(main):003:0> put 'test', 'row1', 'cf1', 'val2'
0 row(s) in 0.0520 seconds
hbase(main):004:0> scan 'test'
ROW COLUMN+CELL
row1 column=cf1:, timestamp=1297853125623, value=val2
1 row(s) in 0.0790 seconds
hbase(main):005:0> scan 'test', { VERSIONS => 3 }
ROW COLUMN+CELL
row1 column=cf1:, timestamp=1297853125623, value=val2
row1 column=cf1:, timestamp=1297853122412, value=val1
1 row(s) in 0.0640 seconds
테이블을 생성하고 동일한 키(rowid-column family- column)를 갖는 값을 두개 put한다.
이러면 rowId-column family-column 에 두개의 버전이 저장된다.
이를 조회하기 위해서는 VERSIONS라는 메타값을 지정한다.
scan 'test', {VERSIONS=>3}
버전을 최대 3개까지 보여달라는 것이다.
그냥 scan 만 한경우에는 버전들 중에서 가장 최근에 저장된 데이터를 보여주게 된다.
그렇다면 동일한 키를 가지고 저장을 100번하면 버전 100개가 저장되는 걸까? 물론 아니다, 최대 저장할 수 있는 버전의 수를 지정할 수 있다. 물론 major-compact 를 수행하기 전에는 보관하고 있을수 있지만 major-compact 에서 MAX-VERSIONS수만 남기고 모두 삭제하는듯하다.
Client Write Buffer
HBase API는 클라이언트측 쓰기 버퍼를 지원한다. 이는 put 오퍼레이션을 한꺼번에 서버에 전달하기 위함이다.
void setAutoFlush(boolean autoFlush)
boolean isAutoFlush()
디폴트로 클라이언트측 쓰기 버퍼는 off되어 있다. 오토 플러쉬를 false로 설정함으로써 버퍼를 활성화 시킬수 있다.
table.setAutoFlush(false);
그럼 언제 서버에게 전달하는가?
flushCommits 라는 함수를 호출하면 리모트서버들에게 변경을 알린다.
long getWriteBufferSize
언제 flush될까? flushCommit 호출, 일정 사이즈를 넘으면 자동으로 (setWriteBufferSize(long writeBufferSize) thows Exceptoin) - 또는 hbase-site.xml 파일에 다음과 같이 지정한다. (2mb), close 함수 호출시 결국 flushCommits이 호출된다.
<property>
<name>hbase.client.write.buffer</name>
<value>20971520</value>
</property
ArrayList<Put> getWriteBuffer
버퍼에 있는 put 인스턴스의 리스트를 리턴한다.
Put 오류 대처
예) 잘못된 Column family를 삽입할려고하는 예제
Put put1 = new Put(Bytes.toBytes("row1"));
put1.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"),
Bytes.toBytes("val1"));
puts.add(put1);
Put put2 = new Put(Bytes.toBytes("row2"));
put2.add(Bytes.toBytes("BOGUS"), Bytes.toBytes("qual1"),
Bytes.toBytes("val2"));
puts.add(put2);
Put put3 = new Put(Bytes.toBytes("row2"));
put3.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual2"),
Bytes.toBytes("val3"));
puts.add(put3);
table.put(puts);
다음과 같은 Exception이 발생한다.
org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException:
Failed 1 action: NoSuchColumnFamilyException: 1 time,
servers with issues: 10.0.0.57:51640,
Hbase 셀에서 삽입 결과를 확인해보면 val1 과 val3이 저장되어 있는것을 확인할 수 있다.
scan 'testtable'
서버측에서는 put1, put2, put3 이 모두 전달되고 모든 오퍼레이션들을 반복적으로 처리하고 오류가 발생하면 각각 리턴하는 형태로 수행된다. Exception 이 발생되면 서버측에서 실패된 Put 인스턴스는 클라이언트측 쓰기버퍼에 유지되고 있다. 그래서 다음번 flush때 재시도 할 것이다. getWriteBuffer를 통해 접근할 수 있다.
클라이언트측에서 에러가 발생한 경우
예)
Put put4 = new Put(Bytes.toBytes("row2"));
puts.add(put4);
table.put(puts);
비어있는 put을 삽입하려고 하면 java.lang.IllegalArgumentException이 발생한다.
Put하기 전에 셀이 조건에 만족하는지 확인한 후 Put하기
boolean checkAndPut(byte[] row, byte[] family, byte[] qualifier,
byte[] value, Put put) throws IOException
예) 해당 셸에 값이 존재하는지 확인하여 없는 경우만 Put하기
주의할점은 동일한 row를 변경하고자 할 때에 적용할 수 있다. 그렇지 않은 경우는 다음과 같은 Exception이 발생한다.
Exception in thread "main" org.apache.hadoop.hbase.DoNotRetryIOException:
Action's getRow must match the passed row