MapReduce实例

Question

我读到了mapreduce，我想知道一个特定的场景。假设我们有几个文件(例如fileA、fileB、fileC )，每个文件由多个整数组成。如果我们想对所有文件中的数字进行排序，以创建如下内容：

23 fileA
34 fileB
35 fileA
60 fileA
60 fileC

地图和减少流程将如何工作？

目前，这是我所拥有的，但并不完全正确；

1. (fileName, fileContent) -> (map to) (Number, fileName)
2. 对临时键、值对进行排序，并获取(Number, (list of){fileName1, fileName2...})
3. 减少临时对并得到 (号码，fileName1) (号码，fileName2) 诸若此类

问题是，在排序阶段，文件名可能不是按字母顺序排列的，因此减少部分无法生成正确的输出。有人能提供一些关于这个场景的正确方法的见解吗？

Answer 1

实现这一目标的最佳方法是通过二次排序。您需要对键(用用例编号)和值(在案例文件名中)进行排序。在Hadoop中，映射器输出仅按键排序。

这可以通过使用复合键来实现:键是数字和文件名的组合。例如，对于第一次记录，键将是(23，fileA)，而不是仅仅(23)。

您可以在这里阅读有关二级排序的内容：https://www.safaribooksonline.com/library/view/data-algorithms/9781491906170/ch01.html

您还可以阅读“备用排序”一节，在"Hadoop的权威指南“一书中。

为了简单起见，我编写了一个程序来实现同样的目的。

在此程序中，默认情况下由映射器对密钥进行排序。我写了一个逻辑来对减速机一侧的值进行排序。因此，它负责对键和值进行排序，并生成所需的输出。

以下是节目内容：

package com.myorg.hadooptests;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.*;

public class SortedValue {


    public static class SortedValueMapper
            extends Mapper<LongWritable, Text , Text, IntWritable>{

        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String[] tokens = value.toString().split(" ");

            if(tokens.length == 2) {
                context.write(new Text(tokens[1]), new IntWritable(Integer.parseInt(tokens[0])));
            }
        }
    }

    public static class SortedValueReducer
            extends Reducer<Text, IntWritable, IntWritable, Text> {

        Map<String, ArrayList<Integer>> valueMap = new HashMap<String, ArrayList<Integer>>();

        public void reduce(Text key, Iterable<IntWritable> values,
                           Context context) throws IOException, InterruptedException {

            String keyStr = key.toString();
            ArrayList<Integer> storedValues = valueMap.get(keyStr);

            for (IntWritable value : values) {
                if (storedValues == null) {
                    storedValues = new ArrayList<Integer>();
                    valueMap.put(keyStr, storedValues);
                }
                storedValues.add(value.get());
            }

            Collections.sort(storedValues);
            for (Integer val : storedValues) {
                context.write(new IntWritable(val), key);
            }
        }
    }

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf, "CompositeKeyExample");
        job.setJarByClass(SortedValue.class);
        job.setMapperClass(SortedValueMapper.class);
        job.setReducerClass(SortedValueReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path("/in/in1.txt"));
        FileOutputFormat.setOutputPath(job, new Path("/out/"));

        System.exit(job.waitForCompletion(true) ? 0:1);

    }
}

Mapper逻辑：

1. 分析每一行。假设键和值由空白字符(“")分隔。
2. 如果该行包含两个令牌，则会发出(文件名，整数值)。例如，对于第一次记录，它发出(fileA，23)。

还原逻辑：

1. 它将(键，值)对放在HashMap中，其中键是文件名，值是该文件的整数列表。例如，对于fileA，存储的值为23、34和35
2. 最后，它对特定键的值和每个值从减速机发出的值(值、键)进行排序。例如，对于fileA，记录输出是：(23，fileA)，(34，fileA)和(35，fileA)

我运行这个程序是为了进行以下输入：

34 fileB
35 fileA
60 fileC
60 fileA
23 fileA

我得到了以下输出：

23      fileA
35      fileA
60      fileA
34      fileB
60      fileC