Hadoop DataJoin软件包使用注意

obeckham

　　第一次使用datajoin软件包进行数据连接，是模仿着《Hadoop实战》那本书写的。解决了几个问题（后面说）之后终于job成功启动并且mapper执行完毕，但reduce总是不能执行完成，进度卡在66.67%或99.33%。mapper输出是想要的格式，而且小规模输入数据测试时整个job能成功完成。在查看未完成reduce的状态，发现key不变，“numOfValues”的值却一直在增大，于是就觉得应该是碰到了死循环。我的reducer中没什么会导致死循环的地方，于是查看DataJoinReducerBase的源码，打印了一些信息之后查看的重点就落在了regroup()方法中，最后终于让我找到了罪魁祸首（下面的红色加粗代码）：



while (arg1.hasNext()) {
this.numOfValues += 1;
if (this.numOfValues % 100 == 0) {
reporter.setStatus("key: " + key.toString() + " numOfValues: "
+ this.numOfValues);
}
if (this.numOfValues > this.maxNumOfValuesPerGroup) {
  continue;
}
aRecord = ((TaggedMapOutput) arg1.next()).clone(job);
Text tag = aRecord.getTag();
ResetableIterator data = retv.get(tag);
if (data == null) {
data = createResetableIterator();
retv.put(tag, data);
}
data.add(aRecord);
}
　　这块代码迭代arg1参数（集合类型的），但这个arg1的大小不能超过this.maxNumOfValuesPerGroup，否则就死循环。因为continue之前并没有从迭代器中取数据的操作，取数据的部分偏偏又执行不到，于是悲剧了。。。
　　其实解决起来也简单，因为



public void configure(JobConf job) {
super.configure(job);
this.job = job;
this.maxNumOfValuesPerGroup = job.getLong("datajoin.maxNumOfValuesPerGroup", 100);
}
　　只要我们在初始化job时，设置datajoin.maxNumOfValuesPerGroup为一个很大的值就好了。具体大小是大于一个group key所对应value的最大个数。不确定到底有多大，干脆设置为Long.MAX_VALUE。
　　实话说，这个属性的默认值较小，在regroup方法中使用方法有点脑残（不知道新版本中改了没有，我用的hadoop-1.0.4）。不过datajoin包是以在内存中缓存记录来实现reduce端连接的，所以内存容量成为它的一个限制，对每个group的数据量有要求。
　　再说说碰到的其他问题：
　　1. 一个单独的mapper类，不是某个类的内部类，在配置Job时也要设置job.setJarByClass();否则会收到classnotfound exception.
　　2. TaggedMapOutput的子类（《Hadoop实战》书中叫TaggedWritable）需要有个无参的构造函数，否则会收到 cannot find function .init（）。这个这本书中也没讲。