hadoop - 如何设计用于合并两个文本文件的 MapReduce 模型

Question

例如我有两个文件，A和B。它们都是文本文件。

A 包含这样一行文本:

I'm A

B 包含如下三行文本:

I'm B1
I'm B2
I'm B3

我的异常(exception)是将A的唯一一行添加到B的每一行的前面，所以结果将是:

I'm A I'm B1
I'm A I'm B2
I'm A I'm B3

对我来说最困难的部分是如何理解 MapReduce 作业的参数。在传统函数中，我可以让 A 和 B 成为两个参数，然后将它们合并到我的规则中。像这样的一些伪代码:

// Two parameters, A and B
public void merge(File A, File B) {
    String lineA = A.firstLine;
    for (Each lineB in B) {
        String result = lineA + " " + lineB;
        File.wirte(result, "ResultFile", appended);
    }
}

但是在MapReduce阶段，没有“传统参数”，例如:

public void map(LongWritable key, Text value, Context context) {
    String line = value.toString();
    try {
        context.write(new Text(line), new Text(""));
    } catch (IOException e) {
        e.printStackTrace();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

我理解上面的代码会逐行读取文件，将每一行作为键，将值设置为空，然后将键值对写入文件。

但同样，上面的映射器代码没有传统参数，我不知道如何执行合并字符串的逻辑，并将结果写出来。

下面是我现在在想的伪代码，我的设计是正确的还是方向错了？我应该如何设计 MapReduce 作业？一些示例代码会很有帮助!

// I think I should pass in two Files
// So I can merge the line of them
public void map(File A, File B, Context context) {
    String key = A.firstLine();
    for (each LineB in B) {
        String value = key + " " + lineB;
    }
    try {
        context.write(new Text(""), value);
    } catch (IOException e) {
        e.printStackTrace();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

Answer 1

这是一个交叉连接——将 A 中的每条记录连接到 B 中的每条记录。幸运的是，这是一个覆盖面很广的 MapReduce 设计模式。

解决方案取决于您的数据集有多大。如果 A 或 B 可以放入内存中，那么你很幸运。只需将它们中的任何一个添加到 DistributedCache 中，只将另一个读入映射器即可。为此，您甚至不需要 reducer !

如果两个数据集都很大，您的解决方案将必须涉及编写自定义输入拆分，以保证每组数据输出足够的次数以匹配另一组的所有 block 。参见 Cross product in MapReduce - 此示例也包含在 MapReduce 设计模式一书中。