hadoop 文件合并

来自：http://blog.csdn.net/dandingyy/article/details/7490046

众所周知，Hadoop对处理单个大文件比处理多个小文件更有效率，另外单个文件也非常占用HDFS的存储空间。所以往往要将其合并起来。

1，getmerge

hadoop有一个命令行工具getmerge，用于将一组HDFS上的文件复制到本地计算机以前进行合并

参考：http://hadoop.apache.org/common/docs/r0.19.2/cn/hdfs_shell.html

使用方法：hadoop fs -getmerge <src> <localdst> [addnl]

接受一个源目录和一个目标文件作为输入，并且将源目录中所有的文件连接成本地目标文件。addnl是可选的，用于指定在每个文件结尾添加一个换行符。

多嘴几句：调用文件系统(FS)Shell命令应使用 bin/hadoop fs <args>的形式。 所有的的FS shell命令使用URI路径作为参数。URI格式是scheme://authority/path。

2.putmerge

将本地小文件合并上传到HDFS文件系统中。

一种方法可以现在本地写一个脚本，先将一个文件合并为一个大文件，然后将整个大文件上传，这种方法占用大量的本地磁盘空间；

另一种方法如下，在复制的过程中上传。参考：《hadoop in action》

[java] view plaincopy

1. import java.io.IOException;
2. import org.apache.hadoop.conf.Configuration;
3. import org.apache.hadoop.fs.FSDataInputStream;
4. import org.apache.hadoop.fs.FSDataOutputStream;
5. import org.apache.hadoop.fs.FileStatus;
6. import org.apache.hadoop.fs.FileSystem;
7. import org.apache.hadoop.fs.Path;
8. import org.apache.hadoop.io.IOUtils;
9. //参数1为本地目录，参数2为HDFS上的文件
10. public class PutMerge {
11. public static void putMergeFunc(String LocalDir, String fsFile) throws IOException
12. {
13. Configuration  conf = new Configuration();
14. FileSystem fs = FileSystem.get(conf);       //fs是HDFS文件系统
15. FileSystem local = FileSystem.getLocal(conf);   //本地文件系统
16. Path localDir = new Path(LocalDir);
17. Path HDFSFile = new Path(fsFile);
18. FileStatus[] status =  local.listStatus(localDir);  //得到输入目录
19. FSDataOutputStream out = fs.create(HDFSFile);       //在HDFS上创建输出文件
20. for(FileStatus st: status)
21. {
22. Path temp = st.getPath();
23. FSDataInputStream in = local.open(temp);
24. IOUtils.copyBytes(in, out, 4096, false);    //读取in流中的内容放入out
25. in.close(); //完成后，关闭当前文件输入流
26. }
27. out.close();
28. }
29. public static void main(String [] args) throws IOException
30. {
31. String l = "/home/kqiao/hadoop/MyHadoopCodes/putmergeFiles";
32. String f = "hdfs://ubuntu:9000/user/kqiao/test/PutMergeTest";
33. putMergeFunc(l,f);
34. }
35. }

3.将小文件打包成SequenceFile的MapReduce任务

来自：《hadoop权威指南》

实现将整个文件作为一条记录处理的InputFormat:

[java] view plaincopy

1. public class WholeFileInputFormat
2. extends FileInputFormat<NullWritable, BytesWritable> {
3. @Override
4. protected boolean isSplitable(JobContext context, Path file) {
5. return false;
6. }
7. @Override
8. public RecordReader<NullWritable, BytesWritable> createRecordReader(
9. InputSplit split, TaskAttemptContext context) throws IOException,
10. InterruptedException {
11. WholeFileRecordReader reader = new WholeFileRecordReader();
12. reader.initialize(split, context);
13. return reader;
14. }
15. }

实现上面类中使用的定制的RecordReader:

[java] view plaincopy

1. /实现一个定制的RecordReader，这六个方法均为继承的RecordReader要求的虚函数。
2. //实现的RecordReader，为自定义的InputFormat服务
3. public class WholeFileRecordReader extends RecordReader<NullWritable, BytesWritable>{
4. private FileSplit fileSplit;
5. private Configuration conf;
6. private BytesWritable value = new BytesWritable();
7. private boolean processed = false;
8. @Override
9. public void close() throws IOException {
10. // do nothing
11. }
12. @Override
13. public NullWritable getCurrentKey() throws IOException,
14. InterruptedException {
15. return NullWritable.get();
16. }
17. @Override
18. public BytesWritable getCurrentValue() throws IOException,
19. InterruptedException {
20. return value;
21. }
22. @Override
23. public float getProgress() throws IOException, InterruptedException {
24. return processed? 1.0f : 0.0f;
25. }
26. @Override
27. public void initialize(InputSplit split, TaskAttemptContext context)
28. throws IOException, InterruptedException {
29. this.fileSplit = (FileSplit) split;
30. this.conf = context.getConfiguration();
31. }
32. //process表示记录是否已经被处理过
33. @Override
34. public boolean nextKeyValue() throws IOException, InterruptedException {
35. if (!processed) {
36. byte[] contents = new byte[(int) fileSplit.getLength()];
37. Path file = fileSplit.getPath();
38. FileSystem fs = file.getFileSystem(conf);
39. FSDataInputStream in = null;
40. try {
41. in = fs.open(file);
42. //将file文件中 的内容放入contents数组中。使用了IOUtils实用类的readFully方法，将in流中得内容放入
43. //contents字节数组中。
44. IOUtils.readFully(in, contents, 0, contents.length);
45. //BytesWritable是一个可用做key或value的字节序列，而ByteWritable是单个字节。
46. //将value的内容设置为contents的值
47. value.set(contents, 0, contents.length);
48. } finally {
49. IOUtils.closeStream(in);
50. }
51. processed = true;
52. return true;
53. }
54. return false;
55. }
56. }

将小文件打包成SequenceFile:

[java] view plaincopy

1. public class SmallFilesToSequenceFileConverter extends Configured implements Tool{
2. //静态内部类，作为mapper
3. static class SequenceFileMapper extends Mapper<NullWritable, BytesWritable, Text, BytesWritable>
4. {
5. private Text filenameKey;
6. //setup在task开始前调用，这里主要是初始化filenamekey
7. @Override
8. protected void setup(Context context)
9. {
10. InputSplit split = context.getInputSplit();
11. Path path = ((FileSplit) split).getPath();
12. filenameKey = new Text(path.toString());
13. }
14. @Override
15. public void map(NullWritable key, BytesWritable value, Context context)
16. throws IOException, InterruptedException{
17. context.write(filenameKey, value);
18. }
19. }
20. @Override
21. public int run(String[] args) throws Exception {
22. Configuration conf = new Configuration();
23. Job job = new Job(conf);
24. job.setJobName("SmallFilesToSequenceFileConverter");
25. FileInputFormat.addInputPath(job, new Path(args[0]));
26. FileOutputFormat.setOutputPath(job, new Path(args[1]));
27. //再次理解此处设置的输入输出格式。。。它表示的是一种对文件划分，索引的方法
28. job.setInputFormatClass(WholeFileInputFormat.class);
29. job.setOutputFormatClass(SequenceFileOutputFormat.class);
30. //此处的设置是最终输出的key/value，一定要注意！
31. job.setOutputKeyClass(Text.class);
32. job.setOutputValueClass(BytesWritable.class);
33. job.setMapperClass(SequenceFileMapper.class);
34. return job.waitForCompletion(true) ? 0 : 1;
35. }
36. public static void main(String [] args) throws Exception
37. {
38. int exitCode = ToolRunner.run(new SmallFilesToSequenceFileConverter(), args);
39. System.exit(exitCode);
40. }
41. }