MapReduce中使用SequenceFile的方式上传文件到集群中

时间：2023-03-09 04:05:17
MapReduce中使用SequenceFile的方式上传文件到集群中

HDFS上不适合存储小文件,因为如果有很多的小文件,上传到HDFS集群,每个文件都会对应一个block块,一个block块的大小默认是128M,对于很多的小文件来说占用了非常多的block数量,就会影响到内存的消耗,

MapReduce处理这些文件的话也是需要很多的Map来处理.

HDFS提供的小文件的解决方案可以使用SequenceFile和MapFile:

如果存在大量的小数据文件，可以使用SequenceFile.

同时使用SequenceFile还可以用SequenceFile自带的一些压缩算法来减少这些细小文件的占用空间.

1.使用SequenceFile相关代码把本地Windows上的很多小文件上传到HDFS集群.

 package seq;

 import java.io.File;

 import java.net.URI;

 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.BytesWritable;

 import org.apache.hadoop.io.SequenceFile;

 import org.apache.hadoop.io.Text;

 public class Test2 {

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

         Configuration conf = new Configuration();

         org.apache.hadoop.fs.FileSystem fs = FileSystem.newInstance(new URI("hdfs://crxy99:9000"),conf);

         Path out = new Path("/members.seq");//输出到HDFS的根目录下"/" 文件命名为memebers.seq

         SequenceFile.Writer writer = new SequenceFile.Writer(fs, conf, out, Text.class, BytesWritable.class);//文件名作为key 类型是Text  文件内容作为值上传上去,类型是BytesWritable

         File localDir = new File("F:\\360Downloads\\crxy\\video\\2016-05-10【mapreduce】 - 副本\\members2000");

         for (File file : localDir.listFiles()) {

             Text key = new Text(file.getName());

             BytesWritable val = new BytesWritable(FileUtils.readFileToByteArray(file));

             writer.append(key, val);

             System.out.println(file.getName());

         }

         writer.close();

     }

 }

程序运行之后查看HDFS目录:

MapReduce中使用SequenceFile的方式上传文件到集群中

通过Web浏览HDFS集群可以看到members.seq文件的大小是126.54MB....只占用一个block.

上传的是一个在Windows本地的members的文件.  Windows本地用户是ABC.

2.使用SequenceFile的block和record压缩算法进行上传文件的相关代码:

 import java.io.File;

 import java.net.URI;

 import org.apache.commons.io.FileUtils;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.fs.FSDataOutputStream;

 import org.apache.hadoop.fs.FileSystem;

 import org.apache.hadoop.fs.Path;

 import org.apache.hadoop.io.BytesWritable;

 import org.apache.hadoop.io.IOUtils;

 import org.apache.hadoop.io.SequenceFile;

 import org.apache.hadoop.io.SequenceFile.CompressionType;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.io.compress.GzipCodec;

 public class Test1 {

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

         Configuration conf = new Configuration();

         org.apache.hadoop.fs.FileSystem fs = FileSystem.newInstance(new URI("hdfs://crxy99:9000"),conf);

         CompressionType type = null;

         if("record".equals(args[0])){

             type = CompressionType.RECORD;

         }

         if("block".equals(args[0])){

             type = CompressionType.BLOCK;

         }

         FSDataOutputStream out = fs.create(new Path(args[1]));

         SequenceFile.Writer writer = SequenceFile.createWriter(conf, out, Text.class, BytesWritable.class,type,new GzipCodec());

         File localDir = new File("/usr/local/hadoop_repo/files/members2000");

         for (File file : localDir.listFiles()) {

             Text key = new Text(file.getName());

             BytesWritable val = new BytesWritable(FileUtils.readFileToByteArray(file));

             writer.append(key, val);

             System.out.println(file.getName());

         }

         writer.close();

         IOUtils.closeStream(out);

     }

 }

结果仍然如上图,文件占用的空间更小.

相关文章