HBase와 MapReduce

HBase와 MapReduce (HBase and MapReduce)

HBase의 데이터로 MapReduce 작업을 실행하는 데 필요한 설정과 상호작용을 설명하는 페이지예요. CLASSPATH 구성, 스캔 캐싱, 예제 코드를 다룹니다.

출처: 문서

본문

Apache MapReduce는 대량의 데이터를 분석하는 데 사용되는 소프트웨어 프레임워크예요. Apache Hadoop에서 제공해요. MapReduce 자체는 이 문서의 범위를 벗어나요. MapReduce를 시작하기 좋은 곳은 https://hadoop.apache.org/docs/r2.6.0/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html이에요. MapReduce 버전 2(MR2)는 이제 YARN의 일부예요.

이 챕터는 HBase 내의 데이터에 MapReduce를 사용하기 위해 취해야 할 특정 구성 단계를 논의해요. 또한 HBase와 MapReduce 작업 사이의 다른 상호작용과 이슈도 논의해요.

mapred와 mapreduce

MapReduce 자체와 마찬가지로 HBase에도 org.apache.hadoop.hbase.mapred와 org.apache.hadoop.hbase.mapreduce 두 개의 mapreduce 패키지가 있어요. 전자는 구식 API를, 후자는 새 방식을 사용해요. 후자가 더 많은 기능을 가지지만 보통 구식 패키지에서 동등한 것을 찾을 수 있어요. 여러분의 MapReduce 배포에 맞는 패키지를 선택하세요. 확신이 없거나 새로 시작한다면 org.apache.hadoop.hbase.mapreduce를 선택하세요. 아래 메모에서 우리는 o.a.h.h.mapreduce를 언급하지만, 사용 중인 것이 있다면 o.a.h.h.mapred로 바꾸세요.

HBase, MapReduce, CLASSPATH (HBase, MapReduce, and the CLASSPATH)

기본적으로 MapReduce 클러스터에 배포된 MapReduce 작업은 $HBASE_CONF_DIR 아래의 HBase 설정이나 HBase 클래스에 접근할 수 없어요.

MapReduce 작업에 필요한 접근을 주려면 _$HADOOP_HOME/conf_에 hbase-site.xml을 추가하고 $HADOOP_HOME/lib 디렉터리에 HBase jar를 추가할 수 있어요. 그런 다음 이 변경 사항을 클러스터 전체에 복사해야 해요. 또는 $HADOOP_HOME/conf/hadoop-env.sh를 편집하고 HADOOP_CLASSPATH 변수에 hbase 의존성을 추가할 수 있어요. 이 두 접근 모두 Hadoop 설치를 HBase 참조로 오염시키기 때문에 권장되지 않아요. 또한 Hadoop이 HBase 데이터를 사용하려면 Hadoop 클러스터를 재시작해야 해요.

권장 접근은 HBase가 의존성 jar를 추가하게 하고 HADOOP_CLASSPATH나 -libjars를 사용하는 것이에요.

HBase 0.90.x부터 HBase는 작업 설정 자체에 의존성 JAR을 추가해요. 의존성은 로컬 CLASSPATH에서 사용 가능하기만 하면 되고, 여기서 집어들어 MapReduce 클러스터에 배포되는 fat 작업 jar에 번들돼요. 기본 트릭은 전체 hbase 클래스패스 — 모든 hbase와 의존 jar 및 설정 — 를 mapreduce 작업 러너에 전달해, hbase 유틸리티가 전체 클래스패스에서 필요한 것을 골라 MapReduce 작업 설정에 추가하게 하는 것이에요(이것이 어떻게 되는지는 TableMapReduceUtil#addDependencyJars(org.apache.hadoop.mapreduce.Job) 소스 참고).

다음 예제는 usertable이라는 테이블에 대해 번들된 HBase RowCounter MapReduce 작업을 실행해요. hbase가 MapReduce 컨텍스트에서 실행하는 데 필요한 jar(hbase-site.xml 같은 설정 파일 포함)를 HADOOP_CLASSPATH에 설정해요. 시스템에 맞는 올바른 HBase JAR 버전을 사용해야 하며, 아래 커맨드라인의 VERSION 문자열을 로컬 hbase 설치 버전으로 바꾸세요. 백틱(`)은 셸이 하위 명령을 실행하게 해서 hbase classpath의 출력을 HADOOP_CLASSPATH에 설정해요. 이 예제는 BASH 호환 셸을 사용한다고 가정해요.

$ HADOOP_CLASSPATH=${HBASE_HOME}/bin/hbase classpath
${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/lib/hbase-mapreduce-VERSION.jar
org.apache.hadoop.hbase.mapreduce.RowCounter usertable

위 명령은 hadoop 설정이 가리키는 클러스터에서 로컬 설정이 가리키는 hbase 클러스터에 대해 행 카운팅 mapreduce 작업을 시작할 거예요.

hbase-mapreduce.jar의 main은 hbase와 함께 제공되는 몇 가지 기본 mapreduce 작업을 나열하는 Driver예요. 예를 들어 설치가 hbase 2.0.0-SNAPSHOT이라고 가정하면:

$ HADOOP_CLASSPATH=${HBASE_HOME}/bin/hbase classpath
${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/lib/hbase-mapreduce-2.0.0-SNAPSHOT.jar An example program must be given as the first argument. Valid program names are: CellCounter: Count cells in HBase table. WALPlayer: Replay WAL files. completebulkload: Complete a bulk data load. copytable: Export a table from local cluster to peer cluster. export: Write table data to HDFS. exportsnapshot: Export the specific snapshot to a given FileSystem. import: Import data written by Export. importtsv: Import data in TSV format. rowcounter: Count rows in HBase table. verifyrep: Compare the data from tables in two different clusters. WARNING: It doesn't work for incrementColumnValues'd cells since the timestamp is changed after being appended to the log.

아래의 행 카운터 작업 재실행에서처럼 위에 나열된 mapreduce 작업용 짧은 이름을 사용할 수 있어요(다시, 설치가 hbase 2.0.0-SNAPSHOT이라고 가정):

$ HADOOP_CLASSPATH=${HBASE_HOME}/bin/hbase classpath
${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/lib/hbase-mapreduce-2.0.0-SNAPSHOT.jar
rowcounter usertable

더 선택적인 hbase mapredcp 도구 출력이 흥미로울 수 있어요. hbase 설치에 대해 기본 mapreduce 작업을 실행하는 데 필요한 최소 jar 집합을 나열해요. 설정은 포함하지 않아요. MapReduce 작업이 대상 클러스터를 찾게 하려면 아마 이것들을 추가해야 할 거예요. 실질적인 작업을 시작하면 추가 jar에 대한 포인터도 추가해야 할 거예요. hbase mapredcp를 실행할 때 시스템 속성 -Dtmpjars를 전달해 추가 항목을 지정하면 돼요.

의존성을 패키징하지 않거나 TableMapReduceUtil#addDependencyJars를 호출하지 않는 작업에는 다음 명령 구조가 필요해요.

$ HADOOP_CLASSPATH=${HBASE_HOME}/bin/hbase mapredcp:${HBASE_HOME}/conf hadoop jar MyApp.jar MyJobMainClass -libjars $(${HBASE_HOME}/bin/hbase mapredcp | tr ':' ',') ...

설치된 위치가 아니라 빌드 디렉터리에서 HBase를 실행한다면 이 예제가 동작하지 않을 수 있어요. 다음과 같은 오류가 보일 수 있어요.

java.lang.RuntimeException: java.lang.ClassNotFoundException: org.apache.hadoop.hbase.mapreduce.RowCounter$RowCounterMapper

이 경우 빌드 환경의 target/ 디렉터리에서 HBase JAR을 사용하도록 다음처럼 명령을 수정해 보세요.

$ HADOOP_CLASSPATH=${HBASE_BUILD_HOME}/hbase-mapreduce/target/hbase-mapreduce-VERSION-SNAPSHOT.jar:${HBASE_BUILD_HOME}/bin/hbase classpath ${HADOOP_HOME}/bin/hadoop jar ${HBASE_BUILD_HOME}/hbase-mapreduce/target/hbase-mapreduce-VERSION-SNAPSHOT.jar rowcounter usertable

0.96.1~0.98.4 사이 HBase의 MapReduce 사용자에게 알림 (Notice to MapReduce users of HBase between 0.96.1 and 0.98.4)

HBase를 사용하는 일부 MapReduce 작업이 시작에 실패해요. 증상은 다음과 유사한 예외예요.

Exception in thread "main" java.lang.IllegalAccessError: class com.google.protobuf.ZeroCopyLiteralByteString cannot access its superclass com.google.protobuf.LiteralByteString at java.lang.ClassLoader.defineClass1(Native Method) at java.lang.ClassLoader.defineClass(ClassLoader.java:792) at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142) at java.net.URLClassLoader.defineClass(URLClassLoader.java:449) at java.net.URLClassLoader.access$100(URLClassLoader.java:71) at java.net.URLClassLoader$1.run(URLClassLoader.java:361) at java.net.URLClassLoader$1.run(URLClassLoader.java:355) at java.security.AccessController.doPrivileged(Native Method) at java.net.URLClassLoader.findClass(URLClassLoader.java:354) at java.lang.ClassLoader.loadClass(ClassLoader.java:424) at java.lang.ClassLoader.loadClass(ClassLoader.java:357) at org.apache.hadoop.hbase.protobuf.ProtobufUtil.toScan(ProtobufUtil.java:818) at org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil.convertScanToString(TableMapReduceUtil.java:433) at org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil.initTableMapperJob(TableMapReduceUtil.java:186) at org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil.initTableMapperJob(TableMapReduceUtil.java:147) at org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil.initTableMapperJob(TableMapReduceUtil.java:270) at org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil.initTableMapperJob(TableMapReduceUtil.java:100) ...

이것은 HBASE-9867에서 도입된, 의도치 않게 클래스로더 의존성을 도입한 최적화 때문이에요. 이는 -libjars 옵션을 사용하는 작업과 중첩된 lib 폴더에 런타임 의존성을 패키징하는 "fat jar" 모두에 영향을 줘요. 새 클래스로더 요구 사항을 충족하려면 hbase-protocol.jar를 Hadoop의 클래스패스에 포함해야 해요. 클래스패스 오류 해결에 대한 현재 권장 사항은 HBase, MapReduce, and the CLASSPATH를 참고하세요. 다음은 역사적 목적으로 포함된 것이에요. 이것은 심링크를 통해 또는 jar를 새 위치로 복사해 hbase-protocol.jar에 대한 참조를 Hadoop의 lib 디렉터리에 포함시켜 시스템 전역으로 해결할 수 있어요. 또한 작업 제출 시점에 HADOOP_CLASSPATH 환경 변수에 포함시켜 작업별 실행 기준으로 달성할 수도 있어요. 의존성을 패키징하는 작업을 시작할 때 다음 세 명령 모두 이 요구 사항을 충족해요.

$ HADOOP_CLASSPATH=/path/to/hbase-protocol.jar:/path/to/hbase/conf hadoop jar MyJob.jar MyJobMainClass $ HADOOP_CLASSPATH=$(hbase mapredcp):/path/to/hbase/conf hadoop jar MyJob.jar MyJobMainClass $ HADOOP_CLASSPATH=$(hbase classpath) hadoop jar MyJob.jar MyJobMainClass

의존성을 패키징하지 않는 jar의 경우 다음 명령 구조가 필요해요.

$ HADOOP_CLASSPATH=$(hbase mapredcp):/etc/hbase/conf hadoop jar MyApp.jar MyJobMainClass -libjars $(hbase mapredcp | tr ':' ',') ...

이 이슈에 대한 추가 논의는 HBASE-10304도 참고하세요.

MapReduce Scan Caching

TableMapReduceUtil은 이제 전달되는 Scan 객체에 스캐너 캐싱(결과를 클라이언트로 반환하기 전에 캐시되는 행 수)을 설정하는 옵션을 복원해요. 이 기능은 HBase 0.95의 버그(HBASE-11558)로 잃었었으며, HBase 0.98.5와 0.96.3에서 수정됐어요. 스캐너 캐싱 선택의 우선순위 순서는 다음과 같아요.

scan 객체에 설정된 캐싱 설정.
설정 옵션 hbase.client.scanner.caching으로 지정된 캐싱 설정. hbase-site.xml에서 수동으로 설정하거나 헬퍼 메서드 TableMapReduceUtil.setScannerCaching()으로 설정할 수 있어요.
기본값 HConstants.DEFAULT_HBASE_CLIENT_SCANNER_CACHING. 100으로 설정돼요.

캐싱 설정을 최적화하는 것은 클라이언트가 결과를 기다리는 시간과 클라이언트가 받아야 하는 결과 집합 수 사이의 균형이에요. 캐싱 설정이 너무 크면 클라이언트가 오래 기다리거나 요청이 타임아웃될 수도 있어요. 설정이 너무 작으면 스캔이 결과를 여러 조각으로 반환해야 해요. 스캔을 삽(shovel)으로 생각하면, 더 큰 캐시 설정은 더 큰 삽에 해당하고, 더 작은 캐시 설정은 버킷을 채우기 위해 더 많은 삽질을 하는 것과 같아요.

위에 언급한 우선순위 목록은 합리적인 기본값을 설정하고 특정 연산에 대해 오버라이드할 수 있게 해줘요.

자세한 내용은 Scan의 API 문서를 참고하세요.

번들된 HBase MapReduce 작업 (Bundled HBase MapReduce Jobs)

HBase JAR은 일부 번들된 MapReduce 작업의 Driver 역할도 해요. 번들된 MapReduce 작업을 알아보려면 다음 명령을 실행하세요.

$ ${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/hbase-mapreduce-VERSION.jar An example program must be given as the first argument. Valid program names are: copytable: Export a table from local cluster to peer cluster completebulkload: Complete a bulk data load. export: Write table data to HDFS. import: Import data written by Export. importtsv: Import data in TSV format. rowcounter: Count rows in HBase table

유효한 프로그램 이름 각각은 번들된 MapReduce 작업이에요. 작업 중 하나를 실행하려면 다음 예제를 본떠 명령을 모델링하세요.

$ ${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/hbase-mapreduce-VERSION.jar rowcounter myTable

MapReduce 작업의 데이터 소스·싱크로서의 HBase (HBase as a MapReduce Job Data Source and Data Sink)

HBase는 MapReduce 작업의 데이터 소스인 TableInputFormat 및 데이터 싱크인 TableOutputFormat 또는 MultiTableOutputFormat으로 사용될 수 있어요. HBase를 읽거나 쓰는 MapReduce 작업을 작성할 때는 TableMapper 및/또는 TableReducer를 서브클래스화하는 것이 좋아요. 기본 사용법은 아무것도 하지 않는 통과 클래스 IdentityTableMapper와 IdentityTableReducer를 참고하세요. 더 관여된 예시는 RowCounter를 보거나 org.apache.hadoop.hbase.mapreduce.TestTableMapReduce 단위 테스트를 검토하세요.

HBase를 소스나 싱크로 사용하는 MapReduce 작업을 실행한다면 설정에 소스·싱크 테이블과 컬럼 이름을 지정해야 해요.

HBase에서 읽을 때 TableInputFormat은 HBase에서 region 목록을 요청하고, map-per-region 또는 mapreduce.job.maps 중 더 작은 맵을 만들어요. 작업에 맵이 두 개뿐이라면 mapreduce.job.maps를 region 수보다 큰 수로 올리세요. 노드당 TaskTracer/NodeManager와 RegionServer를 실행한다면 맵은 인접한 TaskTracker/NodeManager에서 실행될 거예요. HBase에 쓸 때 Reduce 단계를 피하고 맵 내에서 HBase로 다시 쓰는 것이 합리적일 수 있어요. 이 접근은 작업이 MapReduce가 맵이 방출한 데이터에 수행하는 정렬과 콜레이션을 필요로 하지 않을 때 동작해요. 삽입 시 HBase가 '정렬'을 하므로 필요하지 않다면 이중 정렬(그리고 MapReduce 클러스터 주위로 데이터를 셔플링)할 이유가 없어요. Reduce가 필요 없다면 맵이 작업 끝에 보고하기 위해 처리된 레코드 수를 방출하거나, Reduce 수를 0으로 설정하고 TableOutputFormat을 사용할 수 있어요. 여러분의 경우 Reduce 단계를 실행하는 것이 합리적이라면, 보통 여러 개의 reducer를 사용해 부하가 HBase 클러스터 전체에 퍼지도록 해야 해요.

새 HBase 파티셔너인 HRegionPartitioner는 기존 region 수만큼의 reducer를 실행할 수 있어요. HRegionPartitioner는 테이블이 크고 업로드가 완료 시 기존 region 수를 크게 바꾸지 않을 때 적합해요. 그렇지 않으면 기본 파티셔너를 사용하세요.

벌크 임포트 중 HFile 직접 쓰기 (Writing HFiles Directly During Bulk Import)

새 테이블로 임포트한다면 HBase API를 우회하고 내용을 HBase 데이터 파일(HFiles)로 포맷해 파일시스템에 직접 쓸 수 있어요. 임포트가 더 빠르게, 어쩌면 한 단계 더 빠르게 실행될 거예요. 이 메커니즘이 어떻게 동작하는지에 대한 자세한 내용은 Bulk Load를 참고하세요.

RowCounter 예시 (RowCounter Example)

포함된 RowCounter MapReduce 작업은 TableInputFormat을 사용하고 지정된 테이블의 모든 행을 카운트해요. 실행하려면 다음 명령을 사용하세요.

$ ./bin/hadoop jar hbase-X.X.X.jar

이것은 HBase MapReduce Driver 클래스를 호출해요. 제공되는 작업 목록에서 rowcounter를 선택하세요. 그러면 rowcounter 사용법이 표준 출력으로 인쇄될 거예요. 테이블 이름, 카운트할 컬럼, 출력 디렉터리를 지정하세요. 클래스패스 오류가 있으면 HBase, MapReduce, and the CLASSPATH를 참고하세요.

맵 태스크 분할 (Map-Task Splitting)

기본 HBase MapReduce 스플리터 (The Default HBase MapReduce Splitter)

TableInputFormat을 사용해 MapReduce 작업에서 HBase 테이블을 소싱하면, 그 스플리터는 테이블의 각 region마다 맵 태스크를 만들어요. 따라서 테이블에 100개의 region이 있다면, Scan에서 선택된 컬럼 패밀리 수와 무관하게 작업에 100개의 맵 태스크가 있을 거예요.

커스텀 스플리터 (Custom Splitters)

커스텀 스플리터 구현에 관심이 있다면 TableInputFormatBase의 getSplits 메서드를 참고하세요. 맵 태스크 할당 로직이 바로 그곳에 있어요.

HBase MapReduce 예시 (HBase MapReduce Examples)

HBase MapReduce 읽기 예시 (HBase MapReduce Read Example)

다음은 HBase를 읽기 전용 방식으로 MapReduce 소스로 사용하는 예시예요. 구체적으로 Mapper 인스턴스는 있지만 Reducer가 없고, Mapper에서 아무것도 방출되지 않아요. 작업은 다음과 같이 정의될 거예요.

Configuration config = HBaseConfiguration.create(); Job job = new Job(config, "ExampleRead"); job.setJarByClass(MyReadJob.class); // class that contains mapper

Scan scan = new Scan(); scan.setCaching(500); // 1 is the default in Scan, which will be bad for MapReduce jobs scan.setCacheBlocks(false); // don't set to true for MR jobs // set other scan attrs ...

TableMapReduceUtil.initTableMapperJob( tableName, // input HBase table name scan, // Scan instance to control CF and attribute selection MyMapper.class, // mapper null, // mapper output key null, // mapper output value job); job.setOutputFormatClass(NullOutputFormat.class); // because we aren't emitting anything from mapper

boolean b = job.waitForCompletion(true); if (!b) { throw new IOException("error with job!"); }

...그리고 mapper 인스턴스는 TableMapper를 확장할 거예요...

public static class MyMapper extends TableMapper<Text, Text> {

public void map(ImmutableBytesWritable row, Result value, Context context) throws InterruptedException, IOException { // process data for the row from the Result instance. } }

HBase MapReduce 읽기/쓰기 예시 (HBase MapReduce Read/Write Example)

다음은 MapReduce에서 HBase를 소스와 싱크로 모두 사용하는 예시예요. 이 예제는 한 테이블에서 다른 테이블로 데이터를 복사할 거예요.

Configuration config = HBaseConfiguration.create(); Job job = new Job(config,"ExampleReadWrite"); job.setJarByClass(MyReadWriteJob.class); // class that contains mapper

Scan scan = new Scan(); scan.setCaching(500); // 1 is the default in Scan, which will be bad for MapReduce jobs scan.setCacheBlocks(false); // don't set to true for MR jobs // set other scan attrs

TableMapReduceUtil.initTableMapperJob( sourceTable, // input table scan, // Scan instance to control CF and attribute selection MyMapper.class, // mapper class null, // mapper output key null, // mapper output value job); TableMapReduceUtil.initTableReducerJob( targetTable, // output table null, // reducer class job); job.setNumReduceTasks(0);

boolean b = job.waitForCompletion(true); if (!b) { throw new IOException("error with job!"); }

특히 reducer와 관련해 TableMapReduceUtil이 무엇을 하는지 설명이 필요해요. TableOutputFormat이 outputFormat 클래스로 사용되고, config에 여러 매개변수(예: TableOutputFormat.OUTPUT_TABLE)가 설정되며, reducer 출력 키가 ImmutableBytesWritable로, reducer 값이 Writable로 설정돼요. 이들은 프로그래머가 작업과 conf에 설정할 수 있지만, TableMapReduceUtil이 더 쉽게 만들려 해요.

다음은 입력 Result에 맞는 Put을 만들고 방출하는 예제 mapper예요. 메모: 이것이 CopyTable 유틸리티가 하는 일이에요.

public static class MyMapper extends TableMapper<ImmutableBytesWritable, Put> {

public void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException { // this example is just copying the data from the source table... context.write(row, resultToPut(row,value)); }

private static Put resultToPut(ImmutableBytesWritable key, Result result) throws IOException {
  Put put = new Put(key.get());
  for (Cell cell : result.listCells()) {
    put.add(cell);
  }
  return put;
}

}

실제 reducer 단계는 없으므로 TableOutputFormat이 Put을 대상 테이블로 보내는 것을 처리해요.

이것은 단지 예시이며, 개발자는 TableOutputFormat을 사용하지 않고 대상 테이블에 직접 연결할 수도 있어요.

다중 테이블 출력이 있는 HBase MapReduce 읽기/쓰기 예시 (HBase MapReduce Read/Write Example With Multi-Table Output)

TODO: MultiTableOutputFormat 예시.

HBase로 요약하는 HBase MapReduce 예시 (HBase MapReduce Summary to HBase Example)

다음 예제는 요약 단계와 함께 HBase를 MapReduce 소스와 싱크로 사용해요. 이 예제는 테이블에서 값의 고유 인스턴스 수를 세고 그 요약된 카운트를 다른 테이블에 써요.

Configuration config = HBaseConfiguration.create(); Job job = new Job(config,"ExampleSummary"); job.setJarByClass(MySummaryJob.class); // class that contains mapper and reducer

Scan scan = new Scan(); scan.setCaching(500); // 1 is the default in Scan, which will be bad for MapReduce jobs scan.setCacheBlocks(false); // don't set to true for MR jobs // set other scan attrs

TableMapReduceUtil.initTableMapperJob( sourceTable, // input table scan, // Scan instance to control CF and attribute selection MyMapper.class, // mapper class Text.class, // mapper output key IntWritable.class, // mapper output value job); TableMapReduceUtil.initTableReducerJob( targetTable, // output table MyTableReducer.class, // reducer class job); job.setNumReduceTasks(1); // at least one, adjust as required

boolean b = job.waitForCompletion(true); if (!b) { throw new IOException("error with job!"); }

이 예제 mapper에서 String 값을 가진 컬럼이 요약할 값으로 선택돼요. 이 값은 mapper에서 방출할 키로 사용되고, IntWritable은 인스턴스 카운터를 나타내요.

public static class MyMapper extends TableMapper<Text, IntWritable> { public static final byte[] CF = "cf".getBytes(); public static final byte[] ATTR1 = "attr1".getBytes();

private final IntWritable ONE = new IntWritable(1); private Text text = new Text();

public void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException { String val = new String(value.getValue(CF, ATTR1)); text.set(val); // we can only emit Writables... context.write(text, ONE); } }

reducer에서 "1"들이 카운트되고(이를 하는 다른 MR 예제와 마찬가지로), 그다음 Put을 방출해요.

public static class MyTableReducer extends TableReducer<Text, IntWritable, ImmutableBytesWritable> { public static final byte[] CF = "cf".getBytes(); public static final byte[] COUNT = "count".getBytes();

public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { int i = 0; for (IntWritable val : values) { i += val.get(); } Put put = new Put(Bytes.toBytes(key.toString())); put.add(CF, COUNT, Bytes.toBytes(i));

context.write(null, put);

} }

파일로 요약하는 HBase MapReduce 예시 (HBase MapReduce Summary to File Example)

이것은 위의 요약 예제와 매우 유사하며, HBase를 MapReduce 소스로 사용하지만 HDFS를 싱크로 사용한다는 점만 달라요. 차이는 작업 설정과 reducer에 있어요. mapper는 동일하게 유지돼요.

Configuration config = HBaseConfiguration.create(); Job job = new Job(config,"ExampleSummaryToFile"); job.setJarByClass(MySummaryFileJob.class); // class that contains mapper and reducer

Scan scan = new Scan(); scan.setCaching(500); // 1 is the default in Scan, which will be bad for MapReduce jobs scan.setCacheBlocks(false); // don't set to true for MR jobs // set other scan attrs

TableMapReduceUtil.initTableMapperJob( sourceTable, // input table scan, // Scan instance to control CF and attribute selection MyMapper.class, // mapper class Text.class, // mapper output key IntWritable.class, // mapper output value job); job.setReducerClass(MyReducer.class); // reducer class job.setNumReduceTasks(1); // at least one, adjust as required FileOutputFormat.setOutputPath(job, new Path("/tmp/mr/mySummaryFile")); // adjust directories as required

boolean b = job.waitForCompletion(true); if (!b) { throw new IOException("error with job!"); }

위에서 말했듯이, 이전 Mapper는 이 예제에서 변경 없이 실행될 수 있어요. Reducer는 TableMapper를 확장하고 Puts를 방출하는 대신 "일반" Reducer예요.

public static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { int i = 0; for (IntWritable val : values) { i += val.get(); } context.write(key, new IntWritable(i)); } }

Reducer 없이 HBase로 요약하는 HBase MapReduce (HBase MapReduce Summary to HBase Without Reducer)

reducer 없이 요약을 수행하는 것도 가능해요 — HBase를 reducer로 사용한다면요.

작업 요약을 위해 HBase 대상 테이블이 존재해야 해요. Table 메서드 incrementColumnValue가 값을 원자적으로 증가시키는 데 사용될 거예요. 성능 관점에서 각 map-task에 대해 증가시킬 값과 그 값들의 Map을 유지하고, mapper의 cleanup 메서드 중에 키당 한 번 업데이트하는 것이 합리적일 수 있어요. 그러나 처리할 행 수와 고유 키에 따라 결과는 달라질 수 있어요.

결국 요약 결과는 HBase에 있게 돼요.

RDBMS로 요약하는 HBase MapReduce (HBase MapReduce Summary to RDBMS)

때로는 요약을 RDBMS로 생성하는 것이 더 적절할 수 있어요. 그런 경우 커스텀 reducer를 통해 요약을 RDBMS로 직접 생성하는 것이 가능해요. setup 메서드는 RDBMS에 연결할 수 있고(연결 정보는 context의 커스텀 매개변수로 전달될 수 있어요) cleanup 메서드는 연결을 닫을 수 있어요.

작업의 reducer 수가 요약 구현에 영향을 미치며, 이를 reducer에 설계해야 한다는 것을 이해하는 것이 중요해요. 구체적으로 singleton(하나의 reducer)으로 실행되도록 설계되었는지, 여러 reducer로 실행되도록 설계되었는지요. 둘 중 어느 것도 옳고 그르지 않고, 사용 사례에 달려 있어요. 작업에 더 많은 reducer가 할당될수록 RDBMS에 더 많은 동시 연결이 생성된다는 것을 인식하세요. 이는 확장되지만 어느 지점까지만이에요.

public static class MyRdbmsReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

private Connection c = null;

public void setup(Context context) { // create DB connection... }

public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { // do summarization // in this example the keys are Text, but this is just an example }

public void cleanup(Context context) { // close db connection } }

결국 요약 결과는 여러분의 RDBMS 테이블에 쓰여져요.

MapReduce 작업에서 다른 HBase 테이블 접근 (Accessing Other HBase Tables in a MapReduce Job)

프레임워크가 현재 MapReduce 작업에 하나의 HBase 테이블 입력을 허용하지만, Mapper의 setup 메서드에서 Table 인스턴스를 만들어 룩업 테이블 등으로 다른 HBase 테이블에 접근할 수 있어요.

public class MyMapper extends TableMapper<Text, LongWritable> { private Table myOtherTable;

public void setup(Context context) { // In here create a Connection to the cluster and save it or use the Connection // from the existing table myOtherTable = connection.getTable("myOtherTable"); }

public void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException { // process Result... // use 'myOtherTable' for lookups }

추측 실행 (Speculative Execution)

HBase를 소스로 사용하는 MapReduce 작업에서는 추측 실행(speculative execution)을 끄는 것이 일반적으로 좋아요. 이것은 속성을 통해 작업별로 하거나 전체 클러스터에서 할 수 있어요. 특히 더 오래 실행되는 작업에서 추측 실행은 중복 map-task를 만들어 HBase에 데이터를 이중 기록하게 할 거예요. 아마 원치 않을 거예요.

자세한 내용은 Speculative Execution을 참고하세요.

HBase와 MapReduce