MapReduce性能优化--小文件问题

傷城~ 2022-08-30 01:43 207阅读 0赞

针对MapReduce而言，每一个小文件都是一个Block，都会产生一个InputSplit，最终每一个小文件都会 产生一个map任务，这样会导致同时启动太多的Map任务，Map任务的启动是非常消耗性能的，但是启动了以后执行了很短时间就停止了，因为小文件的数据量太小了，这样就会造成任务执行消耗的时间还没有启动任务消耗的时间多，这样也会影响MapReduce执行的效率。

针对这个问题,解决办法通常是选择一个容器，将这些小文件组织起来统一存储，HDFS提供了两种类型的容器，分别是`SequenceFile`和`MapFile`

*  **SequeceFile**

SequeceFile是Hadoop 提供的一种二进制文件，这种二进制文件直接将<key, value>对序列化到文件中。

一般对小文件可以使用这种文件合并，即将小文件的文件名作为key，文件内容作为value序列化到大文件中。

但是这个文件有一个缺点，就是它需要一个合并文件的过程，最终合并的文件会比较大，并且合并后的文件查看起来不方便，必须通过遍历才能查看里面的每一个小文件。

所以这个SequenceFile 其实可以理解为把很多小文件压缩成一个大的压缩包了。

下面我们来具体看一下如何生成SequenceFile,生成SequenceFile需要开发代码:

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{  
    		//生成SequenceFile文件 
    		write("~/smallFile","/seqFile"); 
    		//读取SequenceFile文件
            read("/seqFile");
        }
        
    	/** * 生成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://bigdata01:9000"); 
    		//获取操作HDFS的对象
    		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://bigdata01:9000");
    		//创建阅读器
    		SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFi Text key = new Text();
    		Text value = new Text();
    		//循环读取数据
    		while(reader.next(key,value)){ 
    			//输出文件名称
    			System.out.print("文件名:"+key.toString()+","); 
    			//输出文件的内容 
    			System.out.println("文件内容:"+value.toString());
    		}
    		reader.close();
    	}
    }

*  **MapFile**

MapFile是排序后的SequenceFile, MapFile由两部分组成，分别是`index`和`data`

index作为文件的数据索引，主要记录了每个Record的key值，以及该Record在文件中的偏移位置。在 MapFile被访问的时候,索引文件会被加载到内存，通过索引映射关系可迅速定位到指定Record所在文件位置，

因此，相对SequenceFile而言，MapFile的检索效率是高效的，缺点是会消耗一部分内存来存储index数据。

代码实现如下:

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.MapFile;
    import org.apache.hadoop.io.SequenceFile;
    import org.apache.hadoop.io.Text;
    
    import java.io.File;
    
    /** * 小文件解决方案之MapFile */
    public class SmallFileMap { 
    
    	public static void main(String[] args) throws Exception{  
    		// 生成MapFile文件
    		write("~/smallFile","/mapFile");
    		// 读取MapFile文件
    		read("/mapFile");
    	}
    
    	/** * 生成MapFile文件 * @param inputDir 输入目录-windows目录 * @param outputDir 输出目录-hdfs目录 * @throws Exception */
    	private static void write(String inputDir,String outputDir) throws Exception{ 
    		// 创建一个配置对象
    		Configuration conf = new Configuration(); 
    		// 指定HDFS的地址 
    		conf.set("fs.defaultFS","hdfs://bigdata01:9000"); 
    		//获取操作HDFS的对象
    		FileSystem fileSystem = FileSystem.get(conf);
    		//删除输出目录
    		fileSystem.delete(new Path(outputDir),true);
    		//构造opts数组，有两个元素 
    		/* 第一个是key类型 第二个是value类型 */
    		SequenceFile.Writer.Option[] opts = new SequenceFile.Writer.Option[]{ 
    		        MapFile.Writer.keyClass(Text.class),
    		        MapFile.Writer.valueClass(Text.class)};
    		// 创建一个writer实例
    		MapFile.Writer writer = new MapFile.Writer(conf,new Path(outputDir),o
    		// 指定要压缩的文件的目录
    		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();
    	}
    	
    	/** * 读取MapFile文件 * @param inputDir MapFile文件路径 * @throws Exception */
    	private static void read(String inputDir) throws Exception{ 
    		// 创建一个配置对象
    		Configuration conf = new Configuration();
    		//指定HDFS的地址
    		conf.set("fs.defaultFS","hdfs://bigdata01:9000");
    		// 创建阅读器
    		MapFile.Reader reader = new MapFile.Reader(new Path(inputDir),conf);
    		Text key = new Text();
    		Text value = new Text();
    		// 循环读取数据 
    		while(reader.next(key,value)){ 
    			// 输出文件名称 
    			System.out.print("文件名:"+key.toString()+","); 
    			// 输出文件的内容 
    			System.out.println("文件内容:"+value.toString());
    		}
            reader.close();
        }
    }

执行代码中的write方法，可以看到在HDFS上会产生一个/mapFile目录，这个目录里面有两个文件，一个 index索引文件，一个data数据文件

执行代码中的read方法，可以输出小文件的名称和内容