MapReduce中的输入输出格式控制

在 MapReduce 作业中，可以通过指定 SequenceFileOutputFormat 来生成 SequenceFile 文件。SequenceFile 是 Hadoop 定义的一种二进制文件格式，用于存储大量键值对（key-value）的序列化数据。文件头部包含键和值的数据类型信息。

以下是一个示例，展示了如何在 MapReduce 作业中使用 SequenceFileOutputFormat：

public static class IndexStepOneMapper extends Mapper {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        FileSplit inputSplit = (FileSplit) context.getInputSplit();
        String fileName = inputSplit.getPath().getName();
        String[] words = value.toString().split(" ");
        for (String w : words) {
            context.write(new Text(w + "-" + fileName), new IntWritable(1));
        }
    }
}

public static class IndexStepOneReducer extends Reducer {
    @Override
    protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
        int count = 0;
        for (IntWritable value : values) {
            count += value.get();
        }
        context.write(key, new IntWritable(count));
    }
}

// 设置输出格式为 SequenceFileOutputFormat
job.setOutputFormatClass(SequenceFileOutputFormat.class);

其他步骤与常规的 MapReduce 作业相同。

接下来是第二个步骤，读取前一个步骤生成的 SequenceFile 文件并进行进一步处理：

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
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.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class IndexStepTwo {
    public static class IndexStepTwoMapper extends Mapper {
        @Override
        protected void map(Text key, IntWritable value, Context context) throws IOException, InterruptedException {
            String[] split = key.toString().split("-");
            context.write(new Text(split[0]), new Text(split[1] + "-->" + value));
        }
    }

    public static class IndexStepTwoReducer extends Reducer {
        @Override
        protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
            StringBuilder sb = new StringBuilder();
            for (Text value : values) {
                sb.append(value.toString()).append("\t");
            }
            context.write(key, new Text(sb.toString()));
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration cOnf= new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(IndexStepTwo.class);
        job.setMapperClass(IndexStepTwoMapper.class);
        job.setReducerClass(IndexStepTwoReducer.class);
        job.setNumReduceTasks(1);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setInputFormatClass(SequenceFileInputFormat.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.setInputPaths(job, new Path("F:\mrdata\index\out1"));
        FileOutputFormat.setOutputPath(job, new Path("F:\mrdata\index\out2"));
        job.waitForCompletion(true);
    }
}

以上代码展示了如何在 MapReduce 作业中使用 SequenceFileInputFormat 和 SequenceFileOutputFormat，以实现数据的高效读写和处理。