关系代数选择运算MapReduce并行化 输入数据自己造，不能用别人的，每行一个记录，包括学号，姓名，年龄，班级，要求查找所有18岁的记录。 给出自己造的数据，Mapper，Reducer，Driver三个类，以及运行结果。

数据：

001,Amy,18,ClassA
002,Bob,19,ClassB
003,Charlie,17,ClassA
004,David,18,ClassC
005,Ellen,20,ClassB
006,Frank,18,ClassA
007,Grace,16,ClassC
008,Henry,18,ClassB
009,Ivy,18,ClassA
010,Jack,21,ClassC

Mapper类：

import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class SelectMapper extends Mapper<LongWritable, Text, Text, Text> {

    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] fields = line.split(",");
        String age = fields[2];
        if (age.equals("18")) {
            context.write(new Text(fields[0]), new Text(line));
        }
    }
}

Reducer类：

import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class SelectReducer extends Reducer<Text, Text, Text, Text> {

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {
        for (Text value : values) {
            context.write(null, value);
        }
    }
}

Driver类：

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

public class SelectDriver {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "select");
        job.setJarByClass(SelectDriver.class);
        job.setMapperClass(SelectMapper.class);
        job.setReducerClass(SelectReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

运行结果：

001,Amy,18,ClassA
004,David,18,ClassC
006,Frank,18,ClassA
008,Henry,18,ClassB
009,Ivy,18,ClassA