在实际开发中，我们肯定希望提高MapReduce的工作效率，其实提高MapReduce的效率，无非就是提高Map阶段和Reduce阶段的效率。
Map阶段优化之小文件问题
我们知道Map阶段中的MapTask个数是与InputSplit的个数有关的，一般一个InputSplit切片对应一个，而且InputSplit的个数我们一般也无法控制，应为默认就是128MB，但是往往我们的文件并不是这样，而是大小不一，有的可能300MB，一个可能只有10KB，尤其是为一群几十KB的小文件一个划分一个InputSplit切片，实在浪费资源。
而且针对HDFS而言，每一个小文件在namenode中都会占用150字节的内存空间，最终会导致集群中虽然存储了很多个文件，但是文件的体积并不大，这样就没有意义了。
这就想到我之前的文章【自定义InputFormat】，我们可以这样来将多个小文件都合并到一个文件当中，进入map()方法后仍然需要我们进行处理才能使用，但这并不是标准的Sequence文件，它的输出键是一个文件名，值是我们文件的字节码，所以如果我们要实现一个WordCount，还需要进一步需要使用的话还需要针对每一行的值将字节码转为Stering，在做分词处理。而下面是真正生成一个Sequence文件的代码：
生成Sequence文件
我们需要将三个小文件合并成一个Sequence文件。
import org.apache.commons.io.FileUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.SequenceFile; import org.apache.hadoop.io.Text; import org.apache.hadoop.io.compress.DefaultCodec; import java.io.File; /** * 小文件解决方案之SequenceFile */ public class SmallFileSeq { public static void main(String[] args) throws Exception{ //在hdfs中生成SequenceFile文件到临时目录下 // write("D:\\smallFile","/tmp/seqFile"); //读取SequenceFile文件 // read("/tmp/seqFile"); //在windows端生成SequenceFile文件 // write("D:\\MapReduce_Data_Test\\myinputformat\\input","D:\\MapReduce_Data_Test\\myinputformat\\outputSeq"); write("D:\\MapReduce_Data_Test\\sequence\\word\\","D:\\MapReduce_Data_Test\\sequence\\wordSeq"); //读取SequenceFile文件 注意是文件地址 不是目录地址 // read("D:\\MapReduce_Data_Test\\myinputformat\\outputSeq"); } /** * 生成SequenceFile文件 * @param inputDir 输入目录-windows目录 * @param outputFile 输出文件-hdfs文件 * @throws Exception */ private static void write(String inputDir,String outputFile) throws Exception{ //创建一个配置对象 Configuration conf = new Configuration(); //指定HDFS的地址 // conf.set("fs.defaultFS","hdfs://hadoop102:9000"); //获取操作文件系统对象 FileSystem fileSystem = FileSystem.get(conf); //删除输出文件 fileSystem.delete(new Path(outputFile),true); //构造opts数组，有三个元素 /* 第一个是输出路径 第二个是key类型 第三个是value类型 */ SequenceFile.Writer.Option[] opts = new SequenceFile.Writer.Option[]{ SequenceFile.Writer.file(new Path(outputFile)), SequenceFile.Writer.keyClass(Text.class), SequenceFile.Writer.valueClass(Text.class)}; //创建一个writer实例 SequenceFile.Writer writer = SequenceFile.createWriter(conf, opts); //指定要压缩的文件的目录 File inputDirPath = new File(inputDir); if(inputDirPath.isDirectory()){ File[] files = inputDirPath.listFiles(); for (File file : files) { //获取文件全部内容 String content = FileUtils.readFileToString(file, "UTF-8"); //文件名作为key Text key = new Text(file.getName()); //文件内容作为value Text value = new Text(content); writer.append(key,value); } } writer.close(); } /** * 读取SequenceFile文件 * @param inputFile SequenceFile文件路径 注意是文件地址不是目录地址 * @throws Exception */ private static void read(String inputFile) throws Exception{ //创建一个配置对象 Configuration conf = new Configuration(); //指定HDFS的地址 // conf.set("fs.defaultFS","hdfs://hadoop102:9000"); // 指定windows端的文件路径，注意在Windows系统中，文件系统没有端口号的概念。因此，只需要使用file://协议指定文件路径即可连接本地文件系统。 conf.set("fs.defaultFS","file:///"); //创建阅读器 SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(new Path(inputFile))); Text key = new Text(); Text value = new Text(); //循环读取数据 while(reader.next(key,value)){ //输出文件名称 System.out.print("文件名:"+key.toString()+","); //输出文件的内容 System.out.println("文件内容:\n"+value.toString()); } reader.close(); } }
输入：
输出：
我们可以看到一共会输出两个文件，下面的 outputSeq是真正的Sequence文件 ，文件名是我们自己指定的，而上面的.outputSeq.crc则是一个校验码文件。
接下来要做的就是如何写一个MapReduce程序将我们的小文件内容从这个合并后的Sequence文件中读取出来
Mapper类
我们需要在Job中指定输入格式InputFormat为SequenceFileInputFormat，我们读取Sequence文件只需要指定键和值都为Text即可。
import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.SequenceFile; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.util.StringUtils; import java.io.IOException; public class SeqMapper extends Mapper<Text, Text,Text,LongWritable> { private Text OUT_KEY = new Text(); private LongWritable OUT_VALUE = new LongWritable(1); @Override protected void map(Text key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); System.out.println(line); String[] words = StringUtils.split(line,'\t'); for (String word : words) { System.out.println("key:" + word); OUT_KEY.set(word); context.write(OUT_KEY,OUT_VALUE); } } }
Reducer类
依旧和之前的WordCount没什么两样。
import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class SeqReducer extends Reducer<Text, LongWritable,Text,LongWritable> { private LongWritable OUT_VALUE = new LongWritable(); @Override public void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException { long sum = 0; for (LongWritable val : values) { sum += val.get(); } OUT_VALUE.set(sum); context.write(key,OUT_VALUE); } }
Runner类
与之前不同的是修改输入格式为SequenceFileInputFormat，而且我们读取Sequence文件时，需要指定键和值的类型都为Text类型。
这个值并不是BytesWritable类型，如果设置为BytesWritable会报错：org.apache.hadoop.io.Text cannot be cast to org.apache.hadoop.io.BytesWritable。这就说明Sequence文件读取时的值类型为Text类型。
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.LongWritable; 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.input.SequenceFileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; public class SeqRunner extends Configured implements Tool { public static void main(String[] args) throws Exception { ToolRunner.run(new Configuration(),new SeqRunner(),args); } @Override public int run(String[] args) throws Exception { //1.获取job Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "sequence file word count"); //2.配置jar包路径 job.setJarByClass(SeqRunner.class); //3.关联mapper和reducer job.setMapperClass(SeqMapper.class); job.setReducerClass(SeqReducer.class); //设置输入格式为 Sequence文件格式 job.setInputFormatClass(SequenceFileInputFormat.class); //4.设置map、reduce输出的k、v类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(LongWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); //5.设置统计文件输入的路径,将命令行的第一个参数作为输入文件的路径 FileInputFormat.setInputPaths(job,new Path("D:\\MapReduce_Data_Test\\sequence\\input")); //6.设置结果数据存放路径,将命令行的第二个参数作为数据的输出路径 FileOutputFormat.setOutputPath(job,new Path("D:\\MapReduce_Data_Test\\sequence\\output")); return job.waitForCompletion(true) ? 0 : 1;//verbose：是否监控并打印job的信息 } }
