java - 从 MapReduce 映射器中的输入文件获取唯一行号

Question

我在 HDFS 中复制了一个包含 1000 万行的文件。需要在mapper中处理行号5000到500000。我应该怎么做？

我尝试覆盖 mapper 中的 run() 方法并在那里尝试一个计数器。但是当文件被拆分并且多个映射器在运行时，当然有多个计数器在运行。所以它没有帮助。粘贴下面的代码。

@Override
public void run(Mapper<LongWritable, Text, Text, Text>.Context context)
        throws IOException, InterruptedException {

    setup(context);

    Integer counter = 0;
    while (context.nextKeyValue()) {

        LongWritable currentKey = context.getCurrentKey();
        Text currentValue = context.getCurrentValue();

        System.out.println(currentKey.toString());

        map(currentKey, currentValue, context);
        counter++;
    }

    System.out.println("Counter: " + counter + " Time: "
            + System.currentTimeMillis());
}

此外，我在映射器中获得的 KEY 不是行号，而是行的偏移量。我们可以得到指向行号的 KEY 吗？如果是这样，它在多个映射器中是否是唯一的？ (当前 KEY，偏移量，在映射器中不是唯一的)。

我怎样才能做对？

Answer 1

• 默认的 InputFormats 例如 TextInputFormat 会给出记录的字节偏移量而不是实际的行号——这主要是因为当一个输入文件是可分割的并且被处理时无法确定真实的行号两个或更多映射器。

• 您可以创建自己的 InputFormat 来生成行号而不是字节偏移量，但您需要将输入格式配置为从 isSplittable 方法返回 false(大型输入文件不会被多个映射器处理)。如果您有小文件，或者文件大小接近 HDFS block 大小，那么这不是问题。

• 您还可以使用 pig 来清理数据并获取那些特别感兴趣的行并处理该特定数据。

  我觉得这是 Hadoop 的缺点，当你想跨不同系统共享全局状态时，Hadoop 会失败。