[hbase] 제한적인 트랙잭션

Hbase에서는 single row operation단위(한 region server에서 한 single row 단위로)로 transaction이 제한적으로 가능하다. 관련 내용을 공부했다. 

1.   MultiRowMutationProtocol.mutateRows()  이용

출처 :

http://hadoop-hbase.blogspot.kr/2012_02_01_archive.html

http://www.slideshare.net/cloudera/1-serving-apparel-catalog-from-h-base-suraj-varma-gap-inc-finalupdated-last-minute

The key observation is that all that is required for atomic operations are co-located cells.

HBASE-5304(https://issues.apache.org/jira/browse/HBASE-5304), HBASE-5368(https://issues.apache.org/jira/browse/HBASE-5368), HBASE-5229(https://issues.apache.org/jira/browse/HBASE-5229)을 통해서 prefix row key을 동일한 이름을 주어  한 Region에 co-located시킬 수 있고  atomic operation을 할 수 있게 한다. 

관련 API

http://hbase.apache.org/apidocs/org/apache/hadoop/hbase/regionserver/IncreasingToUpperBoundRegionSplitPolicy.html

http://hbase.apache.org/apidocs/org/apache/hadoop/hbase/regionserver/DelimitedKeyPrefixRegionSplitPolicy.html

http://hbase.apache.org/apidocs/org/apache/hadoop/hbase/regionserver/KeyPrefixRegionSplitPolicy.html

http://hadoop-hbase.blogspot.kr/2012_02_01_archive.html 의 구현 예제 내용을 그대로 배낌

An example can be setup as follows:

1. add this to hbase-site.xml:
  <property>
    <name>hbase.coprocessor.user.region.classes</name>
    <value>org.apache.hadoop.hbase.coprocessor.MultiRowMutationEndpoint</value>
  </property>

This loads the necessary coprocessor endpoint into all regions of all user tables.

2. Create a table that uses KeyPrefixRegionSplitPolicy:
HTableDescriptor myHtd = new HTableDescriptor("myTable");

myHtd.setValue(HTableDescriptor.SPLIT_POLICY,
               KeyPrefixRegionSplitPolicy.class.getName()); 
// set the prefix to 3 in this example
myHtd.setValue(KeyPrefixRegionSplitPolicy.PREFIX_LENGTH_KEY, 
               String.valueOf(3));

HColumnDescriptor hcd = new HColumnDescriptor(...);
myHtd.addFamily(hcd);

HBaseAdmin admin = ...;

admin.createTable(myHtd);

Regions of "myTable" are now split according to KeyPrefixRegionSplitPolicy with a prefix of 3.


3. Execute an atomic multirow transaction: 

List<Mutation> ms = new ArrayList<Mutation>();
Put p = new Put(Bytes.toBytes("xxxabc"));

...
ms.add(p);
Put p1 = new Put(Bytes.toBytes("xxx123"));

...

ms.add(p1);
Delete d = new Delete(Bytes.toBytes("xxxzzz"));

...

ms.add(d);
// get a proxy for MultiRowMutationEndpoint
// Note that the passed row is used to locate the 
// region. Any of the other row keys could have
// been used as well, as long as they identify 
// the same region.
MultiRowMutationProtocol mr = t.coprocessorProxy(
   MultiRowMutationProtocol.class,
   Bytes.toBytes("xxxabc"));
// perform the atomic operation 
mr.mutateRows(ms);

2. HTable.mutateRow(RowMutations)

하나의 Row를 기반으로 put/delete만 모은 작업을 입력된 순서대로 atomic하게 작업(예전에는 increment도 보장을 요청했지만 그 부분은 제외됨. increment는 무조건 단위로 사용해야 함)

HBASE-3584(https://issues.apache.org/jira/browse/HBASE-3584) 와 HBASE-5203(

https://issues.apache.org/jira/browse/HBASE-5203) 

http://hbase.apache.org/apidocs/org/apache/hadoop/hbase/client/RowMutations.html

Performs multiple mutations atomically on a single row. Currently Put and Delete are supported. The mutations are performed in the order in which they were added.

We compare and equate mutations based off their row so be careful putting RowMutations into Sets or using them as keys in Maps.

public void testColumnFamilyDeleteRM() throws Exception { 

        HTableInterface table = hTablePool.getTable(tableName); 

        try { 

            RowMutations rm =new RowMutations(row); 

            Delete delete = new Delete(row); 

            delete.deleteFamily(cf1); 

            rm.add(delete); 

            System.out.println("Added delete of cf1 column family to row mutation"); 

            Put put = new Put(row); 

            put.add(cf1, Bytes.toBytes("c1"), Bytes.toBytes("new_v1")); 

            put.add(cf1, Bytes.toBytes("c11"), Bytes.toBytes("new_v11")); 

            rm.add(put); 

            System.out.println("Added puts of cf1 column family to row mutation"); 

            table.mutateRow(rm); 

            System.out.println("Mutated row"); 

            Result result = table.get(new Get(row)); 

            NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(cf1); 

            Assert.assertNotNull(familyMap);

            Assert.assertEquals(2, familyMap.size()); 

        } finally { 

            table.close(); 

        } 

    }