# HBase and MapReduce
> 贡献者：[BridgetLai](https://github.com/BridgetLai)
Apache MapReduce 是 [Apache Hadoop](https://hadoop.apache.org/) 提供的软件框架,用来进行大规模数据分析.MapReduce 已超出本文档范围,可通过如下文档学习[https://hadoop.apache.org/docs/r2.6.0/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html](https://hadoop.apache.org/docs/r2.6.0/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html). MapReduce version 2 (MR2)目前是[YARN](https://hadoop.apache.org/docs/r2.3.0/hadoop-yarn/hadoop-yarn-site/)的一部分.
本章将讨论在 HBase 中使用 MapReduce 处理数据时需要进行的一些特定配置步骤;另外,还将讨论 HBase 与 MapReduce jobs 之间的交互以及存在的一些问题;最后将介绍 MapReduce 的[替代 API](http://www.cascading.org/):[Cascading](#cascading).
> `mapred` and `mapreduce`
>
> 与 MapReduce 一样,在 HBase 中也有 2 种 mapreduce API 包.
>_org.apache.hadoop.hbase.mapred_ and _org.apache.hadoop.hbase.mapreduce_.
> 前者使用旧式风格的 API,后者采用新的模式.相比于前者,后者更加灵活,你可以在旧式 API 中找到等价的.选择 API 时,请使用 MapReduce 部署时选择的包.如果不知道如何选择或者想从头再来,那就使用 _org.apache.hadoop.hbase.mapreduce_.在接下来的文章中,将使用 _o.a.h.h.mapreduce_ 如果你使用的是 _o.a.h.h.mapred_ 就自行替换.
## 48\. HBase, MapReduce, and the CLASSPATH
默认情况下,部署在 MapReduce 集群中的 MapReduce jobs 没有权限访问`$HBASE_CONF_DIR`路径下的 HBase 配置 或者 HBase classes.
通过以下方式可以为 MapReduce jobs 配置权限.
> 增加 _hbase-site.xml_ 到 _$HADOOP_HOME/conf_
> <br>然后将 HBase jars 添加到 _$HADOOP_HOME/lib_ 目录下
> <br> 最后需要将这些变更拷贝到 Hadoop 集群中所有服务上.
或者
> 编辑 _$HADOOP_HOME/conf/hadoop-env.sh_ 将 HBase 依赖添加到 `HADOOP_CLASSPATH`.
以上配置均不推荐,因为它会让 Hadoop 安装 HBase 的依赖,并且需要重启 Hadoop 集群才能使用 HBase 中的数据.
推荐的方式是 HBase 使用`HADOOP_CLASSPATH` or `-libjars`添加其依赖的 jar 包.
从 HBase `0.90.x`,HBase 添加依赖 jar 包到任务自身配置中. 依赖项只需要在本地`CLASSPATH`可用,然后被打包部署到 MapReduce 集群的 fat job jar 中.一种取巧的方式是传递全量的 HBase classpath(即 hbase,独立的 jars 还有配置)到 mapreduce job 运行器中令 hbase 工具从全量的 classpath 挑选依赖最终配置到 MapReduce job 的配置中(可以查看源码实现`TableMapReduceUtil#addDependencyJars(org.apache.hadoop.mapreduce.Job)`).
下面的例子是在表`usertable`上运行的 HBase 的 MapReduce 任务: 表行数统计任务[RowCounter](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/RowCounter.html).设置在 MapReduce 上下文运行需要的 hbase jars 以及配置文件如 hbase-site.xml 到 `HADOOP_CLASSPATH`. 一定要确保使用了与你的系统相对应的 HBase Jar.替换以下命令中的 VERSION 字段为本地 HBASE 版本. 反引号(\`)使 shell 执行子命令，将`hbase classpath`的输出设置为`HADOOP_CLASSPATH`. 这个例子需要在 Bash-compatible 执行.
```
$ 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
```
以上命令将启动一个运行在本地配置指定的 hbase 集群的 mapreduce 作业,用来统计表行数.这个集群也是 Hadoop 配置指定的.
`hbase-mapreduce.jar` 核心是一个驱动,罗列了 HBASE 装载的一些基础的 MapReduce 任务.例如,假设你安装的是`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 作业能准确找到目标集群，则可能需要添加些配置)。 一旦你开始做任何实质性的事情，你还需要添加额外依赖,这些依赖需在运行`hbase mapredcp`时通过传递系统属性`-Dtmpjars`来指定。
对于那些没有打包依赖的 jobs 或者直接调用`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 jars
```
$ 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
```
> 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](https://issues.apache.org/jira/browse/HBASE-9867)无意间增加了一个类加载器依赖引入的优化.
> 这个影响使用`-libjars` 和 'fat jar '的任务,他们将运行时依赖放在在`lib`路径下.
为了满足新类加载器需要,`hbase-protocol.jar`必须包含在 Hadoop 的 环境变量下.可通过[HBase, MapReduce, and the CLASSPATH](#hbase.mapreduce.classpath)查阅解决 一些 classpath 错误的推荐解决方法.
The following is included for historical purposes.
>
> 在 Hadoop 的 lib 目录里通过系统连接或者直接拷贝方式引入`hbase-protocol.jar`,可以系统范围内解决 classpath 问题.
>
>这也可以在每个作业启动的基础上实现，方法是在作业提交时将其(`hbase-protocol.jar`)包含在`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](https://issues.apache.org/jira/browse/HBASE-10304)进行更深入的讨论.
## 49\. MapReduce Scan Caching
现在 TableMapReduceUtil 恢复在传入的 Scan 对象上设置扫描器缓存（在将结果返回到客户端之前缓存的行数）的选项。由于 HBase 0.95 中的错误（[HBASE-11558]），此功能丢失 （https://issues.apache.org/jira/browse/HBASE-11558）），修正了 HBase 0.98.5 和 0.96.3。 选择扫描程序缓存的优先顺序如下：
1.在扫描对象上设置的缓存设置。
2.通过配置选项`hbase.client.scanner.caching`指定的缓存设置，可以在 _hbase-site.xml_ 中手动设置，也可以通过辅助方法`TableMapReduceUtil.setScannerCaching（）`设置。
3.默认值`HConstants.DEFAULT_HBASE_CLIENT_SCANNER_CACHING`，设置为“100”。
优化缓存设置是平衡客户端等待结果的时间与客户端需要接收的结果集数量。 如果缓存设置太大，客户端可能会等待很长时间，甚至可能超时。 如果设置太小，则需要将结果分多个部分返回。 如果您将扫描看作是铲子，那么更大的缓存设置相当于更大的铲子，而更小的缓存设置等价于为了填满桶而进行更多的铲动。
上面提到的优先级列表允许您设置一个合理的默认值，也可以根据需要重写。
有关[Scan](https://hbase.apache.org/apidocs/org/apache/hadoop op/hbase/client/scan.html)的更多细节，请参阅 API 文档。
## 50\.捆绑的 HBase MapReduce 作业
HBase JAR 还可用作某些捆绑 MapReduce 作业的驱动程序。要了解捆绑的 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
```
## 51\. HBase 作为 MapReduce 作业数据源和数据接收器
HBase 可以被用作 MapReduce Job 的数据源 [TableInputFormat](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/TableInputFormat.html), 和数据接收器[TableOutputFormat](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/TableOutputFormat.html) or [MultiTableOutputFormat](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/MultiTableOutputFormat.html).编写对 HBase 读写的 MapReduce jbos 时,建议使用子类 [TableMapper](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/TableMapper.html) 或者 [TableReducer](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/TableReducer.html). 有关基本用法请参阅不做任何处理的传递类 [IdentityTableMapper](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/IdentityTableMapper.html) 和 [IdentityTableReducer](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/IdentityTableReducer.html) . 对于更复杂的例子, 请参阅 [RowCounter](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/RowCounter.html) 或者查看 `org.apache.hadoop.hbase.mapreduce.TestTableMapReduce` 的单元测试.
如果运行使用 HBase 作为源或接收器的 MapReduce job，则需要在配置中指定源和接收的表和列名称。
当您从 HBase 读取时，`TableInputFormat`从 HBase 请求分区列表并生成一个映射，该映射可以是“map-per-region”或“mapreduce.job.maps”映射，以较小者为准。如果您的 job 只有两个 maps，请将`mapreduce.job.maps`提升到大于分区数的数字。如果您每个节点正在运行 TaskTracer / NodeManager 和 RegionServer，则 Maps 将在相邻的 TaskTracker / NodeManager 上运行。写入 HBase 时，避免 Reduce 步骤并从 Map 中写回 HBase 可能是有意义的。当您的作业不需要 MapReduce 对 map 发出的数据进行排序和整理时，此方法有效。
在插入时，对于 HBase'排序',除非您需要，否则没有必要对您的 MapReduce 集群进行双重排序（以及对数据进行移动）。如果您不需要 Reduce，您的 map 可能会在作业结束时输出为了报告而处理的记录数，或者将 Reduced 数设置为零并使用 TableOutputFormat。如果在您的情况下运行 Reduce 步骤是有意义的，您通常应该使用多个 reducer，以便负载分布在 HBase 集群中。
新的 HBase 分区程序[HRegionPartitioner]（https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/HRegionPartitioner.html）可以尽可能多的减少运行分区的数量。当您的表很大时，HRegionPartitioner 是合适的，并且您的上传在完成后不会大大改变现有区域的数量。否则请使用默认分区程序。
## 52\. 批量导入时直接写 HFiles
如果要导入新表，则可以绕过 HBase API 并将内容直接写入文件系统，格式化为 HBase 数据文件（HFiles）。您的导入将运行得更快，可能会快一个数量级。有关此机制如何工作的更多信息，请参阅[Bulk Loading](#arch.bulk.load)。
## 53\. 行数统计的例子
包含 [RowCounter](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/RowCounter.html) 的 MapReduce 作业使用 `TableInputFormat` 并且 统计了指定表格的行数.请使用以下命令运行:
```
$ ./bin/hadoop jar hbase-X.X.X.jar
```
这将调用 HBase MapReduce 驱动程序类。请从提供的 jobs 中选择`rowcounter`。这将打印 rowcounter 使用建议到标准输出。指定表名，要计数的列和输出目录。如果您有类路径错误，请参阅 [HBase, MapReduce, and the CLASSPATH](#hbase.mapreduce.classpath).
## 54\. Map 任务 拆分
### 54.1\. HBase 默认的 MapReduce 拆分器
当[TableInputFormat](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/TableInputFormat.html)在 MapReduce job 中被用做获取 HBase 表时,它的拆分器将为表的每个分区指定一个 map 任务.因此如果表中有 100 个分区,无论要扫描多少列,都会为该任务 拆分出 100 个 map 任务.
### 54.2\. 自定义拆分器
如果对自定义拆分器感兴趣,请参阅[TableInputFormatBase](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/TableInputFormatBase.html)中的`getSplits`方法,它是 map 任务拆分逻辑所在.
## 55\. HBase MapReduce 示例
### 55.1\. HBase MapReduce 读示例
以下是以只读方式使用 HBase 作为 MapReduce 源的示例。 具体来说，有一个 Mapper 实例但没有 Reducer，并且 Mapper 没有发出任何内容。 Job 将定义如下......
```
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](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/TableMapper.html)…
```
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.
}
}
```
### 55.2\. HBase MapReduce 读/写示例
以下 HBase 既作为源也作为 MapReduce 的接收器的示例。 此示例将简单地将数据从一个表复制到另一个表。
```
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!");
}
```
很有必要解释一下`TableMapReduceUtil`的作用是什么,尤其是 reducer. [TableOutputFormat](https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/TableOutputFormat.html) 被用作 outputFormat class ,一些参数已经进行了配置,例如`TableOutputFormat.OUTPUT_TABLE`,同时设置了 reducer 的 output key 为`TableOutputFormat.OUTPUT_TABLE` 并且 value 为`Writable`. 这些配置项可以由开发工程师在 job 和配置文件中进行设置,`TableMapReduceUtil`试图将这些工作进行简化.
下面是一个 mapper 的例子,它将创建一个"Put" ,匹配输入的"Result "并输出.而这些工作正是 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 (KeyValue kv : result.raw()) {
put.add(kv);
}
return put;
}
}
```
这实际上并不是一个 reducer 过程, 所以由`TableOutputFormat` 负责将'Put'发送到目标表.
这只是一个例子,开发人员可以选择不使用`TableOutputFormat`并自行链接到目标表.
### 55.3\. HBase MapReduce 多表输出的读写示例
TODO: `MultiTableOutputFormat` 样例.
### 55.4\. HBase MapReduce 汇总到 HBase 示例
以下示例使用 HBase 作为 MapReduce 源并接收汇总信息。此示例将计算表中某个 value 的不同实例的数量，并将这些汇总计数写入另一个表中。
```
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 类型的 value 上进行汇总操作,并将 value作为 mapper 输出的 key,`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 中，计算“ones”（就像执行此操作的任何其他 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<IntWritable> 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);
}
}
```
### 55.5\. HBase MapReduce 文件汇总示例
这与上面的汇总示例很相似,不同之处在于该汇总使用 HBase 作为 MapReduce 的数据源而使用 HDFS 作为接收器.这样的不同体现在 job 启动和 reduce 过程,而 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!");
}
```
如上所述,本示例的中的 mappper 与上例无异,至于 Reducer,则采用一个'通用'的而不是继承自 TableMapper 并且发出 Puts.
```
public static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
int i = 0;
for (IntWritable val : values) {
i += val.get();
}
context.write(key, new IntWritable(i));
}
}
```
### 55.6 不使用 Reducer ,HBase MapReduce 汇总到 HBase
如果使用 HBase 作为 Reducer,也可以在没有 Reducer 的情况下进行汇总.
汇总任务要求 HBase 目标表存在.表方法`incrementColumnValue`将被用作值的原子增长.从性能角度看,为每个 map-task 中那些会值增长的值保留一个 Map,并且在 mapper 执行`cleanup` 方法时每个 key 更新一次,这可能是有意义的.但是，您的里程可能会根据要处理的行数和惟一键的不同而有所不同。
最后,汇总结果在 HBase 中.
### 55.7\. HBase MapReduce 汇总到 RDBMS
有时，为 RDBMS 生成摘要更合适。对于这些情况，可以通过自定义 reducer 直接生成 RDBMS 的摘要。 `setup`方法可以连接到 RDBMS（连接信息可以通过上下文中的自定义参数传递），清理方法可以关闭连接。
一个 job 的 reducer 数量对汇总实现至关重要,您将必须将其设计到 reducer 中.具体来说,不管被设计成一个 reducer 还是多个 reducer,这没有对错之分,完全依赖于您的用例.指定给 job 的 reducer 越多,与 RDMS 建立的实时链接越多,这可以在一定程度上提高吞吐量.
```
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<IntWritable> 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
}
}
```
最终,汇总结果写入到 RDMS 表中.
## 56\. 在 MapReduce 任务中访问其他 HBase 表
尽管目前框架允许一个 HBase 表作为 MapReduce 作业的输入,但其他 HBase 表只可以通过作为查找表(lookup tables)才能访问,例如在 MapReduce 作业中通过 mapper 的 setup 方法创建 Table 实例.
```
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
}
```
## 57\. 推测执行(Speculative Execution)
通常建议关闭以 HBASE 为数据源的 MapReduce 作业的推测执行(speculative execution).关闭推测执行可以通过设置单个任务的属性,也可以设置整个集群.特对是对于执行时间较长的任务,推测执行(speculative execution)为其创建一个重复任务,将进行双重数据写入,这可能不是你想要的.
查阅 [spec.ex](#spec.ex) 获取更多信息.
## 58\. 级联(Cascading)
[[Cascading]（http://www.cascading.org/）是 MapReduce 的替代 API，本质上使用 MapReduce，但允许您以简化的方式编写 MapReduce 代码。
下面的示例显示了一个 Cascading“Flow”，它将数据“汇入”到 HBase 集群中。 同样的`hBaseTap` API 也可用于“获取”数据。
```
// read data from the default filesystem
// emits two fields: "offset" and "line"
Tap source = new Hfs( new TextLine(), inputFileLhs );
// store data in an HBase cluster
// accepts fields "num", "lower", and "upper"
// will automatically scope incoming fields to their proper familyname, "left" or "right"
Fields keyFields = new Fields( "num" );
String[] familyNames = {"left", "right"};
Fields[] valueFields = new Fields[] {new Fields( "lower" ), new Fields( "upper" ) };
Tap hBaseTap = new HBaseTap( "multitable", new HBaseScheme( keyFields, familyNames, valueFields ), SinkMode.REPLACE );
// a simple pipe assembly to parse the input into fields
// a real app would likely chain multiple Pipes together for more complex processing
Pipe parsePipe = new Each( "insert", new Fields( "line" ), new RegexSplitter( new Fields( "num", "lower", "upper" ), " " ) );
// "plan" a cluster executable Flow
// this connects the source Tap and hBaseTap (the sink Tap) to the parsePipe
Flow parseFlow = new FlowConnector( properties ).connect( source, hBaseTap, parsePipe );
// start the flow, and block until complete
parseFlow.complete();
// open an iterator on the HBase table we stuffed data into
TupleEntryIterator iterator = parseFlow.openSink();
while(iterator.hasNext())
{
// print out each tuple from HBase
System.out.println( "iterator.next() = " + iterator.next() );
}
iterator.close();
```
- HBase™ 中文参考指南 3.0
- Preface
- Getting Started
- Apache HBase Configuration
- Upgrading
- The Apache HBase Shell
- Data Model
- HBase and Schema Design
- RegionServer Sizing Rules of Thumb
- HBase and MapReduce
- Securing Apache HBase
- Architecture
- In-memory Compaction
- Backup and Restore
- Synchronous Replication
- Apache HBase APIs
- Apache HBase External APIs
- Thrift API and Filter Language
- HBase and Spark
- Apache HBase Coprocessors
- Apache HBase Performance Tuning
- Troubleshooting and Debugging Apache HBase
- Apache HBase Case Studies
- Apache HBase Operational Management
- Building and Developing Apache HBase
- Unit Testing HBase Applications
- Protobuf in HBase
- Procedure Framework (Pv2): HBASE-12439
- AMv2 Description for Devs
- ZooKeeper
- Community
- Appendix