WordCount 与 Apache Crunch 进入 HBase 独立答案

【问题标题】：WordCount with Apache Crunch into HBase StandaloneWordCount 与 Apache Crunch 进入 HBase 独立
【发布时间】：2014-12-17 14:57:15
【问题描述】：

目前我正在评估 Apache Crunch。我跟着一个简单的WordCount MapReduce job example：之后我尝试将结果保存到独立的 HBase 中。 HBase 正在运行（使用 jps 和 HBase shell 检查），如下所述：http://hbase.apache.org/book/quickstart.html

现在我采用写入HBase的例子：

Pipeline pipeline = new MRPipeline(WordCount.class,getConf());
PCollection<String> lines = pipeline.readTextFile(inputPath);
PTable<String,Long> counts = noStopWords.count();
pipeline.write(counts, new HBaseTarget("wordCountOutTable");
PipelineResult result = pipeline.done();

我得到一个异常：“异常：java.lang.illegalArgumentException：HBaseTarget 只支持放置和删除”

有什么线索吗？

【问题讨论】：

标签： java hadoop mapreduce hbase apache-crunch

【解决方案1】：

PTable 可能是 PCollection，但 HBaseTarget 只能处理 Put 或 Delete 对象。因此，您必须将 PTable 转换为 PCollection，其中集合的每个元素都是 Put 或 Delete。看看Crunch-Examples 是在哪里完成的。

转换示例如下所示：

 public PCollection<Put> createPut(final PTable<String, String> counts) {
   return counts.parallelDo("Convert to puts", new DoFn<Pair<String, String>, Put>() {
     @Override
     public void process(final Pair<String, String> input, final Emitter<Put> emitter) {
       Put put;
       // input.first is used as row key
       put = new Put(Bytes.toBytes(input.first())); 
       // the value (input.second) is added with its family and qualifier
       put.add(COLUMN_FAMILY_TARGET, COLUMN_QUALIFIER_TARGET_TEXT, Bytes.toBytes(input.second())); 
       emitter.emit(put);
     }
   }, Writables.writables(Put.class));
 }

【讨论】：