使用新 API 的 Hadoop 自定义输入格式

Question

我是 Hadoop 的新手，遇到以下问题。我要做的是将数据库的一个分片（请不要问我为什么需要这样做等）映射到映射器，然后对这些数据进行某些操作，将结果输出到减速器并使用该输出再次使用相同的分片格式对相同的数据执行第二阶段映射/缩减作业。 Hadoop 不提供任何输入方法来发送数据库的分片。您只能使用LineInputFormat 和LineRecordReader 逐行发送。 NLineInputFormat 在这种情况下也无济于事。我需要扩展FileInputFormat 和RecordReader 类来编写我自己的InputFormat。有人建议我使用LineRecordReader，因为底层代码已经处理了FileSplits 以及与拆分文件相关的所有问题。 我现在需要做的就是重写 nextKeyValue() 方法，我不知道该怎么做。

       for(int i=0;i<shard_size;i++){
               if(lineRecordReader.nextKeyValue()){
lineValue.append(lineRecordReader.getCurrentValue().getBytes(),0,lineRecordReader.getCurrentValue().getLength());
  }
}

上面的代码sn-p是写出来的，但不知何故不能正常工作。

Answer 1

我建议在您的输入文件中放入连接字符串和其他一些指示可以在哪里找到分片。
Mapper 将获取此信息，连接到数据库并完成工作。我不建议将结果集转换为 hadoop 的可写类 - 这会影响性能。
我认为要解决的问题是对这个相对较小的输入进行足够的拆分。 您可以简单地创建足够的小文件，每个文件都有几个分片引用，或者您可以调整输入格式以构建小拆分。第二种方式会更灵活。

Answer 2

我所做的就是这样。我编写了自己的记录阅读器，一次读取 n 行并将它们作为输入发送给映射器

public boolean nextKeyValue() throws IOException, 

中断异常{

    StringBuilder sb = new StringBuilder();
    for (int i = 0; i < 5; i++) {
        if (!lineRecordReader.nextKeyValue()) {
            return false;
        }
        lineKey = lineRecordReader.getCurrentKey();
        lineValue = lineRecordReader.getCurrentValue();
        sb.append(lineValue.toString());
        sb.append(eol);
    }
    lineValue.set(sb.toString());
    //System.out.println(lineValue.toString());

    return true;

    // throw new UnsupportedOperationException("Not supported yet.");
} 

你是怎么瘦的