Apache HBase Coprocessors

HBase의 고급 기능인 Coprocessor를 설명하는 페이지예요. 데이터가 있는 RegionServer에서 직접 비즈니스 계산을 수행해 네트워크 병목을 피하게 해주는 강력한 확장 메커니즘이에요. 시스템 개발자를 위한 고급 기능입니다.

출처: 문서

본문

HBase Coprocessor는 Google BigTable의 coprocessor 구현(http://research.google.com/people/jeff/SOCC2010-keynote-slides.pdf 41-42쪽)을 모델로 했어요. HBase 구현과 BigTable 아키텍처 사이의 격차를 메우기 위한 노력이 진행 중이에요. 자세한 내용은 HBASE-4047을 참고하세요.

이 챕터의 정보는 주로 다음 자료에서 가져와 크게 재사용했어요:

Mingjie Lai의 블로그 글 Coprocessor Introduction.
Gaurav Bhardwaj의 블로그 글 The How To Of HBase Coprocessors.

Coprocessor는 책임지고 사용하세요 (Use Coprocessors At Your Own Risk)

Coprocessor는 HBase의 고급 기능이며 시스템 개발자만 사용하도록 의도됐어요. Coprocessor 코드는 RegionServer에서 직접 실행되고 데이터에 직접 접근하므로, 데이터 손상, 중간자(man-in-the-middle) 공격, 기타 악의적인 데이터 접근의 위험을 도입해요. 현재 Coprocessor에 의한 데이터 손상을 방지할 메커니즘은 없으며, HBASE-4047에서 작업이 진행 중이에요.

또한 리소스 격리가 없으므로, 선의지만 잘못 동작하는 coprocessor가 클러스터 성능과 안정성을 심각하게 저하시킬 수 있어요.

Coprocessor 개요 (Coprocessor Overview)

HBase에서 Get이나 Scan으로 데이터를 가져와요. 반면 RDBMS에서는 SQL 쿼리를 사용하죠. 관련 데이터만 가져오기 위해 HBase Filter로 필터링하는 반면, RDBMS에서는 WHERE 술어를 사용해요.

데이터를 가져온 뒤 그 데이터에 대해 계산을 수행해요. 이 패러다임은 수천 행과 여러 컬럼을 가진 "작은 데이터"에는 잘 동작해요. 하지만 수십억 행과 수백만 컬럼으로 확장하면, 네트워크를 통해 많은 양의 데이터를 이동하는 것이 네트워크 계층에서 병목을 만들고, 클라이언트가 많은 데이터와 계산을 처리할 만큼 강력하고 메모리가 충분해야 해요. 또한 클라이언트 코드가 커지고 복잡해질 수 있어요.

이 시나리오에서 coprocessor가 의미가 있을 수 있어요. 비즈니스 계산 코드를 데이터와 같은 위치인 RegionServer에서 실행되는 coprocessor에 넣고, 결과를 클라이언트에 반환할 수 있어요.

이것은 coprocessor 사용이 이점을 제공하는 한 시나리오일 뿐이에요. 다음은 coprocessor의 이점 일부를 설명하는 데 도움이 되는 몇 가지 비유예요.

Coprocessor 비유 (Coprocessor Analogies)

트리거와 저장 프로시저 (Triggers and Stored Procedure)

Observer coprocessor는 특정 이벤트(예: Get이나 Put)가 발생하기 전이나 후에 여러분의 코드를 실행한다는 점에서 RDBMS의 트리거와 유사해요. endpoint coprocessor는 클라이언트가 아닌 RegionServer 자체에서 데이터에 대한 커스텀 계산을 수행할 수 있게 해준다는 점에서 RDBMS의 저장 프로시저와 유사해요.

MapReduce

MapReduce는 계산을 데이터의 위치로 이동시키는 원리로 동작해요. Coprocessor도 같은 원리로 동작해요.

AOP

Aspect Oriented Programming(AOP)에 익숙하다면, coprocessor를 요청을 가로채 일부 커스텀 코드를 실행한 뒤 요청을 최종 목적지로 전달(또는 목적지를 변경하기도)함으로써 advice를 적용하는 것으로 생각할 수 있어요.

Coprocessor 구현 개요 (Coprocessor Implementation Overview)

여러분의 클래스는 Coprocessor, RegionObserver, CoprocessorService 등 Coprocessor 인터페이스 중 하나를 구현해야 해요.
구성에서 정적으로 또는 HBase Shell을 사용해 동적으로 coprocessor를 로드하세요. 자세한 내용은 Loading Coprocessors를 참고하세요.
클라이언트 코드에서 coprocessor를 호출하세요. HBase가 coprocessor를 투명하게 처리해요.

프레임워크 API는 coprocessor 패키지에 제공돼요.

Coprocessor 유형 (Types of Coprocessors)

Observer Coprocessors

Observer coprocessor는 특정 이벤트가 발생하기 전이나 후에 트리거돼요. 이벤트 전에 발생하는 observer는 prePut 같은 pre 접두사로 시작하는 메서드를 사용해요. 이벤트 직후에 발생하는 observer는 postPut 같은 post 접두사로 시작하는 메서드를 오버라이드해요.

Observer Coprocessor의 사용 사례 (Use Cases for Observer Coprocessors)

보안 (Security)

Get이나 Put 연산을 수행하기 전에 preGet이나 prePut 메서드로 권한을 확인할 수 있어요.

참조 무결성 (Referential Integrity)

HBase는 RDBMS의 참조 무결성(외래 키라고도 함) 개념을 직접 지원하지 않아요. Coprocessor로 그러한 무결성을 강제할 수 있어요. 예를 들어 users 테이블에 대한 모든 삽입 뒤에 user_daily_attendance 테이블의 해당 항목이 반드시 따라와야 한다는 비즈니스 규칙이 있다면, user에 대한 prePut 메서드를 사용해 user_daily_attendance에 레코드를 삽입하는 coprocessor를 구현할 수 있어요.

보조 인덱스 (Secondary Indexes)

Coprocessor로 보조 인덱스를 유지관리할 수 있어요. 자세한 내용은 SecondaryIndexing을 참고하세요.

Observer Coprocessor 유형 (Types of Observer Coprocessor)

RegionObserver

RegionObserver coprocessor는 Get과 Put 연산 같은 region의 이벤트를 관찰할 수 있게 해줘요. RegionObserver를 참고하세요.

RegionServerObserver

RegionServerObserver는 시작, 중지, merge·commit·rollback 수행 같은 RegionServer 동작과 관련된 이벤트를 관찰할 수 있게 해줘요. RegionServerObserver를 참고하세요.

MasterObserver

MasterObserver는 테이블 생성, 삭제, 스키마 수정 같은 HBase Master와 관련된 이벤트를 관찰할 수 있게 해줘요. MasterObserver를 참고하세요.

WalObserver

WalObserver는 Write-Ahead Log(WAL)에 대한 쓰기와 관련된 이벤트를 관찰할 수 있게 해줘요. WALObserver를 참고하세요.

Examples는 observer coprocessor의 동작 예시를 제공해요.

Endpoint Coprocessor

Endpoint coprocessor는 데이터의 위치에서 계산을 수행할 수 있게 해줘요. Coprocessor Analogy를 참고하세요. 수백 개의 region에 걸친 전체 테이블에 대한 이동 평균(running average)이나 합계를 계산해야 하는 경우가 예시예요.

observer coprocessor가 코드를 투명하게 실행하는 것과 달리, endpoint coprocessor는 AsyncTable에서 사용 가능한 CoprocessorService() 메서드로 명시적으로 호출해야 해요.

sync 클라이언트에서 coprocessorService 메서드를 사용할 때 (On using coprocessorService method with sync client)

Table의 coprocessorService 메서드는 deprecated됐어요. HBASE-21512에서 우리는 async 클라이언트를 기반으로 sync 클라이언트를 재구현했어요. Table 인터페이스에 정의된 coprocessorService 메서드는 protobuf의 BlockingInterface 메서드를 직접 참조하는데, 이는 우리가 async 클라이언트에서 블로킹 호출을 피하고 싶으므로 메서드를 실행하기 위해 별도의 스레드 풀을 사용해야 한다는 뜻이에요. Coprocessor는 고급 기능이므로 coprocessor 사용자가 대신 AsyncTable로 전환하는 것이 괜찮다고 생각해요. 필요하면 Connection에서 AsyncConnection을 얻는 가벼운 toAsyncConnection 메서드가 있어요.

HBase 0.96부터 endpoint coprocessor는 Google Protocol Buffers(protobuf)로 구현돼요. protobuf에 대한 자세한 내용은 Google의 Protocol Buffer Guide를 참고하세요. 0.94 버전에서 작성된 Endpoints Coprocessor는 0.96 이상과 호환되지 않아요. HBASE-5448)을 참고하세요. HBase 클러스터를 0.94 이하에서 0.96 이상으로 업그레이드하려면 coprocessor를 다시 구현해야 해요.

HBase 2.x에서는 protobuf 3.x의 shaded 버전을 사용했지만 coprocessor용 protobuf는 2.5.0으로 유지했어요. HBase 3.0.0에서 우리는 non-shaded protobuf에 대한 모든 의존성을 제거했으므로, hbase-thirdparty에서 제공하는 shaded protobuf 버전을 사용하도록 coprocessor를 다시 구현해야 해요. 자세한 내용은 protobuf 섹션을 참고하세요.

Coprocessor Endpoints는 HBase 내부를 사용해서는 안 되고 공개 API만 사용해야 해요. 이상적으로 CPEP는 인터페이스와 데이터 구조에만 의존해야 해요. 이것이 항상 가능한 것은 아니지만, 그렇게 하면 HBase 내부가 진화함에 따라 Endpoint가 부서지기 쉬워질 수 있다는 점을 경계하세요. private 또는 evolving으로 주석 처리된 HBase 내부 API는 시맨틱 버전 관리 규칙이나 제거 전 deprecation에 대한 일반적인 java 규칙을 따르지 않아도 돼요. 생성된 protobuf 파일에는 hbase audience 주석이 없지만 — 그것들은 HBase가 어떻게 동작하는지 모르는 protobuf protoc 도구로 만들어지므로 — @InterfaceAudience.Private로 간주되어 변경되기 쉽다는 점을 명심해야 해요.

Examples는 endpoint coprocessor의 동작 예시를 제공해요.

Coprocessor 로드 (Loading Coprocessors)

Coprocessor를 HBase에서 사용할 수 있게 하려면 정적으로(HBase 구성으로) 또는 동적으로(HBase Shell이나 Java API로) 로드되어야 해요.

정적 로드 (Static Loading)

Coprocessor를 정적으로 로드하려면 다음 단계를 따르세요. 정적으로 로드된 coprocessor를 내리려면 HBase를 재시작해야 한다는 점을 기억하세요.

hbase-site.xml에 <name>과 <value> 하위 요소가 있는 <property> 요소로 Coprocessor를 정의하세요. <name>은 다음 중 하나여야 해요: RegionObserver와 Endpoint용 hbase.coprocessor.region.classes, WALObserver용 hbase.coprocessor.wal.classes, MasterObserver용 hbase.coprocessor.master.classes. <value>는 coprocessor 구현 클래스의 fully-qualified 클래스 이름을 포함해야 해요. 예를 들어 SumEndPoint.java 클래스에 구현된 Coprocessor를 로드하려면 RegionServer의 'hbase-site.xml' 파일(보통 'conf' 디렉터리 아래)에 다음 항목을 만들어야 해요.

hbase.coprocessor.region.classes org.myname.hbase.coprocessor.endpoint.SumEndPoint

로드할 여러 클래스를 지정하면 클래스 이름을 쉼표로 구분해야 해요. 프레임워크는 기본 클래스 로더로 구성된 모든 클래스를 로드하려 시도해요. 따라서 jar 파일은 서버 측 HBase 클래스패스에 있어야 해요. 이렇게 로드된 Coprocessor는 모든 테이블의 모든 region에서 활성화돼요. 이를 시스템 Coprocessor라고도 불러요. 가장 먼저 나열된 Coprocessor에는 Coprocessor.Priority.SYSTEM 우선순위가 할당돼요. 목록의 각 후속 coprocessor는 우선순위 값이 1씩 증가해요(우선순위는 Integer의 자연 정렬 순서를 가지므로 우선순위가 낮아져요). 이 우선순위 값은 hbase-site.xml에서 수동으로 오버라이드할 수 있어요. 이는 coprocessor가 다른 것 이후에 실행되도록 보장하고 싶을 때 유용해요. 예를 들어 다음 설정에서 SumEndPoint는 다른 coprocessor와 타이를 제외하면 마지막에 실행되도록 보장돼요.

hbase.coprocessor.region.classes org.myname.hbase.coprocessor.endpoint.SumEndPoint|2147483647

등록된 observer를 호출할 때 프레임워크는 우선순위의 정렬 순서로 콜백 메서드를 실행해요. 타이는 임의로 해소돼요.

코드를 HBase의 클래스패스에 넣으세요. 쉬운 방법 중 하나는 HBase 설치의 lib/ 디렉터리에 jar(코드와 모든 의존성 포함)를 넣는 것이에요.
HBase를 재시작하세요.

정적 언로드 (Static Unloading)

hbase-site.xml에서 하위 요소를 포함한 coprocessor의 <property> 요소를 삭제하세요.
HBase를 재시작하세요.
선택적으로, 클래스패스나 HBase의 lib/ 디렉터리에서 coprocessor의 JAR 파일을 제거하세요.

동적 로드 (Dynamic Loading)

HBase를 재시작하지 않고 coprocessor를 동적으로 로드할 수도 있어요. 이것이 정적 로드보다 선호될 것 같지만, 동적으로 로드된 coprocessor는 테이블별로 로드되며 로드된 테이블에만 사용 가능해요. 이런 이유로 동적으로 로드된 테이블은 때로 Table Coprocessor라고 불러요.

또한 coprocessor를 동적으로 로드하는 것은 테이블의 스키마 변경으로 작용하며, coprocessor를 로드하려면 테이블을 오프라인으로 전환해야 해요.

Coprocessor를 동적으로 로드하는 방법은 세 가지가 있어요.

가정 (Assumptions)

아래 지침은 다음을 가정해요:

coprocessor.jar라는 JAR에 모든 의존성과 함께 Coprocessor 구현이 들어 있다.
JAR가 HDFS의 어딘가, 예를 들어 hdfs://NAMENODE:PORT/user/HADOOP_USER/coprocessor.jar에 있다.

HBase Shell 사용 (Using HBase Shell)

다음과 같은 명령으로 Coprocessor를 로드하세요.

hbase alter 'users', METHOD => 'table_att', 'Coprocessor'=>'hdfs://NAMENODE:PORT/user/HADOOP_USER/coprocessor.jar|org.myname.hbase.Coprocessor.RegionObserverExample|1073741823|arg1=1,arg2=2'

Coprocessor 프레임워크는 coprocessor 테이블 속성 값에서 클래스 정보를 읽으려 시도해요. 값은 파이프(|) 문자로 구분된 네 가지 정보를 포함해요.

파일 경로: Coprocessor 구현을 포함한 jar 파일은 모든 region server가 읽을 수 있는 위치에 있어야 해요. 파일을 각 region server의 로컬 디스크에 복사할 수도 있지만, HDFS에 저장하는 것을 권장해요. HBASE-14548은 jar를 포함한 디렉터리나 와일드카드를 지정할 수 있게 해줘요. 예: hdfs://NAMENODE:PORT/user/HADOOP_USER/ 또는 hdfs://NAMENODE:PORT/user/HADOOP_USER/*.jar. 디렉터리를 지정하면 디렉터리의 모든 jar 파일(.jar)이 추가된다는 점에 주의하세요. 서브 디렉터리의 파일은 검색하지 않아요. 디렉터리를 지정하려면 와일드카드를 사용하지 마세요. 이 개선은 JAVA API를 통한 사용에도 적용돼요.
클래스 이름: Coprocessor의 전체 클래스 이름.
우선순위: 정수. 프레임워크가 같은 훅에 등록된 모든 설정된 observer의 실행 순서를 우선순위로 결정해요. 이 필드는 비워둘 수 있어요. 그 경우 프레임워크가 기본 우선순위 값을 할당해요.
인자(선택): 이 필드는 Coprocessor 구현에 전달돼요. 선택 사항이에요.

coprocessor가 로드됐는지 확인하세요.

hbase(main):04:0> describe 'users'

coprocessor는 TABLE_ATTRIBUTES에 나열되어야 해요.

Java API 사용 (모든 HBase 버전) (Using the Java API (all HBase versions))

다음 Java 코드는 HTableDescriptor의 setValue() 메서드를 사용해 users 테이블에 coprocessor를 로드하는 방법을 보여줘요.

TableName tableName = TableName.valueOf("users"); String path = "hdfs://:/user//coprocessor.jar"; Configuration conf = HBaseConfiguration.create(); Connection connection = ConnectionFactory.createConnection(conf); Admin admin = connection.getAdmin(); HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName); HColumnDescriptor columnFamily1 = new HColumnDescriptor("personalDet"); columnFamily1.setMaxVersions(3); hTableDescriptor.addFamily(columnFamily1); HColumnDescriptor columnFamily2 = new HColumnDescriptor("salaryDet"); columnFamily2.setMaxVersions(3); hTableDescriptor.addFamily(columnFamily2); hTableDescriptor.setValue("COPROCESSOR$1", path + "|"

RegionObserverExample.class.getCanonicalName() + "|"
Coprocessor.PRIORITY_USER); admin.modifyTable(tableName, hTableDescriptor);

Java API 사용 (HBase 0.96+ 전용) (Using the Java API (HBase 0.96+ only))

HBase 0.96 이상에서 HTableDescriptor의 addCoprocessor() 메서드는 coprocessor를 동적으로 로드하는 더 쉬운 방법을 제공해요.

TableName tableName = TableName.valueOf("users"); Path path = new Path("hdfs://:/user//coprocessor.jar"); Configuration conf = HBaseConfiguration.create(); Connection connection = ConnectionFactory.createConnection(conf); Admin admin = connection.getAdmin(); HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName); HColumnDescriptor columnFamily1 = new HColumnDescriptor("personalDet"); columnFamily1.setMaxVersions(3); hTableDescriptor.addFamily(columnFamily1); HColumnDescriptor columnFamily2 = new HColumnDescriptor("salaryDet"); columnFamily2.setMaxVersions(3); hTableDescriptor.addFamily(columnFamily2); hTableDescriptor.addCoprocessor(RegionObserverExample.class.getCanonicalName(), path, Coprocessor.PRIORITY_USER, null); admin.modifyTable(tableName, hTableDescriptor);

프레임워크가 주어진 Coprocessor를 성공적으로 로드한다는 보장은 없어요. 예를 들어 셸 명령은 특정 위치에 jar 파일이 존재한다는 것도, 주어진 클래스가 실제로 jar 파일에 포함되어 있다는 것도 보장하지 않아요.

동적 언로드 (Dynamic Unloading)

HBase Shell 사용 (Using HBase Shell)

table_att_unset으로 테이블을 alter해 coprocessor를 제거하세요.

hbase> alter 'users', METHOD => 'table_att_unset', NAME => 'coprocessor$1'

HBASE-26524에 도입된 table_remove_coprocessor로 명시적 클래스명을 지정해 테이블을 alter해 coprocessor를 제거하세요.

hbase> alter 'users', METHOD => 'table_remove_coprocessor', CLASSNAME =>
'org.myname.hbase.Coprocessor.RegionObserverExample'

Java API 사용 (Using the Java API)

setValue()나 addCoprocessor() 메서드로 coprocessor 값을 설정하지 않고 테이블 정의를 다시 로드하세요. 이렇게 하면 테이블에 연결된 모든 coprocessor가 제거돼요.

HBase 0.96 이상에서는 대신 HTableDescriptor 클래스의 removeCoprocessor() 메서드를 사용할 수 있어요.

예시 (Examples)

HBase는 Observer Coprocessor용 예시를 포함해요.

더 자세한 예시는 아래에 있어요.

이 예시들은 개인 정보와 급여 정보를 담고 있는 두 개의 컬럼 패밀리 personalDet와 salaryDet를 가진 users라는 테이블을 가정해요. 아래는 users 테이블의 그림 표현이에요.

Users Table

	personalDet			salaryDet
rowkey	name	lastname	dob	gross	net
admin	Admin	Admin		cdickens	Charles
02/07/1812	10000	8000	2000	jverne	Jules
02/08/1828	12000	9000	3000

Observer 예시 (Observer Example)

다음 Observer coprocessor는 users 테이블의 Get이나 Scan에서 사용자 admin의 세부 정보가 반환되지 않도록 방지해요.

RegionCoprocessor, RegionObserver 클래스를 구현하는 클래스를 작성하세요.
클라이언트가 admin 값을 가진 rowkey를 쿼리했는지 확인하도록 preGetOp() 메서드(preGet() 메서드는 deprecated됨)를 오버라이드하세요. 그렇다면 빈 결과를 반환해요. 그렇지 않으면 요청을 평소처럼 처리하세요.
코드와 의존성을 JAR 파일에 넣으세요.
HBase가 찾을 수 있는 HDFS에 JAR을 놓으세요.
Coprocessor를 로드하세요.
테스트할 간단한 프로그램을 작성하세요.

다음은 위 단계의 구현이에요.

public class RegionObserverExample implements RegionCoprocessor, RegionObserver {

private static final byte[] ADMIN = Bytes.toBytes("admin");
private static final byte[] COLUMN_FAMILY = Bytes.toBytes("details");
private static final byte[] COLUMN = Bytes.toBytes("Admin_det");
private static final byte[] VALUE = Bytes.toBytes("You can't see Admin details");

@Override
public Optional<RegionObserver> getRegionObserver() {
  return Optional.of(this);
}

@Override
public void preGetOp(final ObserverContext<RegionCoprocessorEnvironment> e, final Get get, final List<Cell> results)
throws IOException {

    if (Bytes.equals(get.getRow(),ADMIN)) {
        Cell c = CellUtil.createCell(get.getRow(),COLUMN_FAMILY, COLUMN,
        System.currentTimeMillis(), (byte)4, VALUE);
        results.add(c);
        e.bypass();
    }
}

}

preGetOp()를 오버라이드하는 것은 Get 연산에서만 동작해요. 스캔 결과에서 admin 행을 필터링하려면 preScannerOpen() 메서드도 오버라이드해야 해요.

@Override public RegionScanner preScannerOpen(final ObserverContext e, final Scan scan, final RegionScanner s) throws IOException {

Filter filter = new RowFilter(CompareOp.NOT_EQUAL, new BinaryComparator(ADMIN));
scan.setFilter(filter);
return s;

}

이 메서드는 동작하지만 부작용이 있어요. 클라이언트가 스캔에서 필터를 사용했다면 그 필터가 이 필터로 대체돼요. 대신 스캔에서 admin 결과를 명시적으로 제거할 수 있어요.

@Override public boolean postScannerNext(final ObserverContext e, final InternalScanner s, final List results, final int limit, final boolean hasMore) throws IOException { Result result = null; Iterator iterator = results.iterator(); while (iterator.hasNext()) { result = iterator.next(); if (Bytes.equals(result.getRow(), ROWKEY)) { iterator.remove(); break; } } return hasMore; }

Endpoint 예시 (Endpoint Example)

여전히 users 테이블을 사용하는 이 예시는 endpoint coprocessor를 사용해 모든 직원 급여의 합을 계산하는 coprocessor를 구현해요.

서비스를 정의하는 '.proto' 파일을 만드세요.

option java_package = "org.myname.hbase.coprocessor.autogenerated"; option java_outer_classname = "Sum"; option java_generic_services = true; option java_generate_equals_and_hash = true; option optimize_for = SPEED;

message SumRequest { required string family = 1; required string column = 2; }

message SumResponse { required int64 sum = 1 [default = 0]; }

service SumService { rpc getSum(SumRequest) returns (SumResponse); }

위 .proto' 파일에서 Java 코드를 생성하기 위해 protoc 명령을 실행하세요.

$ mkdir src $ protoc --java_out=src ./sum.proto

이것은 Sum.java라는 클래스를 생성해요.

생성된 서비스 클래스를 확장하고 Coprocessor와 CoprocessorService 클래스를 구현하며 서비스 메서드를 오버라이드하는 클래스를 작성하세요.

hbase-site.xml에서 coprocessor를 로드한 뒤 HBase Shell로 같은 coprocessor를 다시 로드하면 두 번째로 로드돼요. 같은 클래스가 두 번 존재하게 되고, 두 번째 인스턴스는 더 높은 ID(따라서 더 낮은 우선순위)를 가져요. 그 결과 중복 coprocessor는 사실상 무시돼요.

public class SumEndPoint extends Sum.SumService implements Coprocessor, CoprocessorService {

private RegionCoprocessorEnvironment env;

@Override
public Service getService() {
    return this;
}

@Override
public void start(CoprocessorEnvironment env) throws IOException {
    if (env instanceof RegionCoprocessorEnvironment) {
        this.env = (RegionCoprocessorEnvironment)env;
    } else {
        throw new CoprocessorException("Must be loaded on a table region!");
    }
}

@Override
public void stop(CoprocessorEnvironment env) throws IOException {
    // do nothing
}

@Override
public void getSum(RpcController controller, Sum.SumRequest request, RpcCallback<Sum.SumResponse> done) {
    Scan scan = new Scan();
    scan.addFamily(Bytes.toBytes(request.getFamily()));
    scan.addColumn(Bytes.toBytes(request.getFamily()), Bytes.toBytes(request.getColumn()));

    Sum.SumResponse response = null;
    InternalScanner scanner = null;

    try {
        scanner = env.getRegion().getScanner(scan);
        List<Cell> results = new ArrayList<>();
        boolean hasMore = false;
        long sum = 0L;

        do {
            hasMore = scanner.next(results);
            for (Cell cell : results) {
                sum = sum + Bytes.toLong(CellUtil.cloneValue(cell));
            }
            results.clear();
        } while (hasMore);

        response = Sum.SumResponse.newBuilder().setSum(sum).build();
    } catch (IOException ioe) {
        ResponseConverter.setControllerException(controller, ioe);
    } finally {
        if (scanner != null) {
            try {
                scanner.close();
            } catch (IOException ignored) {}
        }
    }

    done.run(response);
}

}

Configuration conf = HBaseConfiguration.create(); Connection connection = ConnectionFactory.createConnection(conf); TableName tableName = TableName.valueOf("users"); Table table = connection.getTable(tableName);

final Sum.SumRequest request = Sum.SumRequest.newBuilder().setFamily("salaryDet").setColumn("gross").build(); try { Map<byte[], Long> results = table.coprocessorService( Sum.SumService.class, null, /* start key / null, / end key */ new Batch.Call<Sum.SumService, Long>() { @Override public Long call(Sum.SumService aggregate) throws IOException { BlockingRpcCallback<Sum.SumResponse> rpcCallback = new BlockingRpcCallback<>(); aggregate.getSum(null, request, rpcCallback); Sum.SumResponse response = rpcCallback.get();

            return response.hasSum() ? response.getSum() : 0L;
        }
    }
);

for (Long sum : results.values()) {
    System.out.println("Sum = " + sum);
}

} catch (ServiceException e) { e.printStackTrace(); } catch (Throwable e) { e.printStackTrace(); }

Coprocessor를 로드하세요.
Coprocessor를 호출하는 클라이언트 코드를 작성하세요.

Coprocessor 배포 지침 (Guidelines For Deploying A Coprocessor)

Coprocessor 번들링 (Bundling Coprocessors)

쉽게 배포하기 위해 coprocessor의 모든 클래스를 RegionServer 클래스패스의 단일 JAR로 번들할 수 있어요. 그렇지 않으면 모든 의존성을 RegionServer 클래스패스에 넣어 RegionServer 시작 중에 로드될 수 있게 하세요. RegionServer의 클래스패스는 RegionServer의 hbase-env.sh 파일에 설정돼요.

배포 자동화 (Automating Deployment)

Puppet, Chef, Ansible 같은 도구로 coprocessor용 JAR을 RegionServer 파일시스템의 필요한 위치로 보내고 각 RegionServer를 재시작해 coprocessor 배포를 자동화할 수 있어요. 그러한 설정의 세부 내용은 이 문서의 범위를 벗어나요.

Coprocessor 업데이트 (Updating a Coprocessor)

주어진 coprocessor의 새 버전을 배포하는 것은 비활성화하고 JAR을 교체하고 다시 활성화하는 것만큼 간단하지 않아요. JVM에서는 현재 참조를 모두 삭제하지 않는 한 클래스를 다시 로드할 수 없기 때문이에요. 현재 JVM이 기존 coprocessor에 대한 참조를 가지므로, 교체하기 위해 RegionServer를 재시작해 JVM을 재시작해야 해요. 이 동작은 바뀌지 않을 것으로 예상돼요.

Coprocessor 로깅 (Coprocessor Logging)

Coprocessor 프레임워크는 표준 Java 로깅을 넘어서는 로깅 API를 제공하지 않아요.

Coprocessor 설정 (Coprocessor Configuration)

HBase Shell에서 coprocessor를 로드하고 싶지 않다면, 설정 속성을 hbase-site.xml에 추가할 수 있어요. Using HBase Shell에서 두 인자가 설정됐어요: arg1=1,arg2=2. 이들은 hbase-site.xml에 다음과 같이 추가할 수 있었어요.

arg1 1 arg2 2

그런 다음 다음과 같은 코드로 설정을 읽을 수 있어요.

Get get = new Get(Bytes.toBytes("admin")); Result result = table.get(get); for (Cell c : result.rawCells()) { System.out.println(Bytes.toString(CellUtil.cloneRow(c)) + "==> " + Bytes.toString(CellUtil.cloneFamily(c)) + "{" + Bytes.toString(CellUtil.cloneQualifier(c)) + ":" + Bytes.toLong(CellUtil.cloneValue(c)) + "}"); } Scan scan = new Scan(); ResultScanner scanner = table.getScanner(scan); for (Result res : scanner) { for (Cell c : res.rawCells()) { System.out.println(Bytes.toString(CellUtil.cloneRow(c)) + " ==> " + Bytes.toString(CellUtil.cloneFamily(c)) + " {" + Bytes.toString(CellUtil.cloneQualifier(c)) + ":" + Bytes.toLong(CellUtil.cloneValue(c)) + "}"); } }

Coprocessor 사용 제한 (Restricting Coprocessor Usage)

임의의 사용자 coprocessor를 제한하는 것은 멀티테넌트 환경에서 큰 우려가 될 수 있어요. HBase는 예상된 coprocessor만 실행되도록 보장하는 연속적인 옵션을 제공해요.

hbase.coprocessor.enabled: 모든 coprocessor를 활성화·비활성화해요. 이는 모든 coprocessor를 비활성화하면 일부 보안 공급자가 비활성화되므로 HBase 기능을 제한할 거예요. 영향을 받는 예시 coprocessor는 org.apache.hadoop.hbase.security.access.AccessController예요.
hbase.coprocessor.user.enabled: 테이블(즉, 사용자 coprocessor)에 coprocessor를 로드하는 것을 활성화·비활성화해요.

hbase-site.xml의 다음 조정 가능 항목으로 coprocessor를 정적으로 로드하고 선택적으로 우선순위를 조정할 수 있어요.

hbase.coprocessor.regionserver.classes: region server가 로드하는 coprocessor의 쉼표 구분 목록
hbase.coprocessor.region.classes: RegionObserver와 Endpoint coprocessor의 쉼표 구분 목록
hbase.coprocessor.user.region.classes: 모든 region이 로드하는 coprocessor의 쉼표 구분 목록
hbase.coprocessor.master.classes: master가 로드하는 coprocessor(MasterObserver coprocessor)의 쉼표 구분 목록
hbase.coprocessor.wal.classes: 로드할 WALObserver coprocessor의 쉼표 구분 목록
hbase.coprocessor.abortonerror: coprocessor가 IOError 외의 오류를 내면 coprocessor를 로드한 데몬을 중단(abort)할지 여부. false로 설정되고 access controller coprocessor에 치명적 오류가 발생하면 coprocessor가 우회되므로, 보안 설치에서는 true가 권장돼요. 그러나 사용자 coprocessor에 대해 테이블별로 오버라이드해서 실행 중인 region server를 중단하지 않고 오류 시 언로드되도록 할 수 있어요.
hbase.coprocessor.region.whitelist.paths: org.apache.hadoop.hbase.security.access.CoprocessorWhitelistMasterObserver를 로드하는 사람들을 위한 쉼표 구분 목록으로, coprocessor가 로드될 수 있는 경로를 화이트리스트로 설정하는 데 다음 옵션을 사용할 수 있어요.

클래스패스의 coprocessor는 암시적으로 화이트리스트 처리돼요

모든 coprocessor 경로를 와일드카드 처리하려면 *
전체 파일시스템(예: hdfs://my-cluster/)
FilenameUtils.wildcardMatch로 평가할 와일드카드 경로

메모: 경로는 스킴을 지정할 수도, 지정하지 않을 수도 있어요(예: 파일시스템용 file:///usr/hbase/lib/coprocessors 또는 모든 파일시스템용 /usr/hbase/lib/coprocessors).

Apache HBase Coprocessors