Hadoop DataJoin软件包使用注意

第一次使用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 <SomeClass>.init（）。这个这本书中也没讲。