java - Hadoop 和 MapReduce，如何将从 csv 中提取的行数组发送到映射函数，其中每个数组包含行 x - y；

Question

好吧，我已经阅读了很多有关 Hadoop 和 MapReduce 的内容，也许是因为我不像大多数人那样熟悉迭代器，但我有一个问题，我似乎也找不到直接答案。基本上，据我了解，映射函数是由许多机器和/或内核并行执行的。因此，无论您正在做什么，都不能依赖于程序之前执行的代码来获得任何类型的速度增益。这对我来说非常有效，但我所做的需要我小批量地测试信息。基本上，我需要将 .csv 中的批量行作为 32、64、128 或任何行的数组发送。就像第 0 – 127 行进入 core1 执行映射函数一样，第 128 – 255 行进入 core2 的执行，等等。另外，我需要在函数内将每个批处理的内容作为一个整体提供，就像我向它传递了一个数组一样。我读了一些关于新的 java API 如何允许所谓的“推”和“拉”的内容，并且这允许批量发送内容，但我找不到任何示例代码。我不知道，我将继续研究，我会发布我发现的任何内容，但如果有人知道，请他们在这个帖子中发布。我真的很感激我可能得到的任何帮助。

编辑

如果您可以简单地确保 .csv 的 block 按顺序发送，您可以通过这种方式执行它。我想这也假设mapreduce 中有全局变量。

//** concept not code **//

GLOBAL_COUNTER = 0;
GLOBAL_ARRAY = NEW ARRAY();

map()
{ 

GLOBAL_ARRAY[GLOBAL_COUNTER] = ITERATOR_VALUE;

GLOBAL_COUNTER++;
if(GLOBAL_COUNTER == 127)
{
//EXECUTE TEST WITH AN ARRAY OF 128 VALUES FOR COMPARISON
GLOBAL_COUNTER = 0;
}

}

Answer 1

如果您尝试将 CSV 文件中的大量行放入映射器中，您可能会考虑编写自己的 InputFormat/RecordReader 以及可能自己的 WritableComparable 对象。使用自定义的InputFormat/RecordReader，您将能够根据收到的输入指定如何创建对象并将其传递给映射器。

如果映射器正在执行您想要的操作，但您需要将这些行 block 发送到化简器，请使映射器的输出键对于同一化简函数中您想要的每一行都相同。

默认的 TextInputFormat 将为您的映射器提供输入，如下所示(本例中的键/偏移量只是随机数):

0    Hello World
123  My name is Sam
456  Foo bar bar foo

每一行都将作为键值对读入映射器。只需将您需要的每一行的 key 修改为相同并将其写入输出即可:

0    Hello World
0    My name is Sam
1    Foo bar bar foo

第一次读取reduce函数时，它会收到一个键值对，键为“0”，值为包含“Hello World”和“My name is Sam”的Iterable对象。您将能够使用 Iterable 对象在同一个reduce 方法调用中访问这两个值。

这是一些伪代码:

int count = 0
map (key, value) {
      int newKey = count/2
      context.write(newKey,value)
      count++
}

reduce (key, values) {
    for value in values
        // Do something to each line
}

希望有帮助。 :)