hadoop - map 侧加入MR工作

Question

我正在使用MapSide join以便在MR作业中获取2个不同的文件。

Input File 1:
0,5.3
26,4.9
54,4.9
.
.
.

InputFile 2:
0   Anju,,3.6,IT,A,1.6,0.3
26  Remya,,3.3,EEE,B,1.6,0.3
54  akhila,,3.3,IT,C,1.3,0.3

我的意图是如下替换

Anju,5.3,3.6,IT,A,1.6,0.3
Remya,4.9,3.3,EEE,B,1.6,0.3
akhila,4.9,3.3,IT,C,1.3,0.3

我所做的是

为了获得2个文件作为输入，我使用了2个mappers( MultipleInput)。

2个文件的第一列是文件偏移量。

在第一个 map 中，我发出了第一个col作为键(可写偏移量)，其余部分作为第一个文件的值发出
在第二张 map 中，我还发出了第一个col作为键(可写偏移量)，其余部分作为第二个文件的值发出了

所以在 reducer 中我可以得到

Reducer
key 0
value 5.3
value Anju,S,3.6,IT,A,1.6,0.3

Reducer
key 26
value Remya,,3.3,EEE,B,1.6,0.3
value 4.9

Reducer
key 54
value 4.9
value akhila,,3.3,IT,C,1.3,0.3

我如何 replace重视任何想法？

我的方法是正确的还是应该采用替代性方式？
请提出建议。

Answer 1

您可以使用以下代码替换:

String result = null;
String replacement = null;
for (Text value: values) {
    String valueStr = value.toString();
    if (valueStr.contains(",,")) {
        result = valueStr;
    } else {
        replacement = valueStr;
    }
}
if (result == null || replacement == null) {
    return;
}
result = result.replaceFirst(",,", "," + replacement + ",");
// write result

但这不是MapSide联接。要进行MapSide联接，您应该在每个映射器中(在设置阶段)读取带有替换文件( InputFile 1)，然后在映射阶段将数据与 InputFile 2联接。例:

private Map < Integer, Double > replacements;

@Override
protected void setup(Context context) throws IOException, InterruptedException {
    replacements = new HashMap < Integer, Double > ();
    // read FileInput 1 to replacements
    // ...
}

@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    String[] parts = value.toString().split("\t"); // 0   Anju,,3.6,IT,A,1.6,0.3
    Integer id = Integer.parseInt(parts[0]);
    if (!replacements.containsKey(id)) {
        return; // can't replace
    }
    Double replacement = replacements.get(id);
    String result = parts[1].replaceFirst(",,", "," + replacement + ",");
    // write result to context
}