hadoop文件合並

眾所周知，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》

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

//參數1為本地目錄，參數2為HDFS上的文件
public class PutMerge {
	
	public static void putMergeFunc(String LocalDir, String fsFile) throws IOException
	{
		Configuration  conf = new Configuration();
		FileSystem fs = FileSystem.get(conf);		//fs是HDFS文件系統
		FileSystem local = FileSystem.getLocal(conf);	//本地文件系統
		
		Path localDir = new Path(LocalDir);
		Path HDFSFile = new Path(fsFile);
		
		FileStatus[] status =  local.listStatus(localDir);	//得到輸入目錄
		FSDataOutputStream out = fs.create(HDFSFile);		//在HDFS上創建輸出文件
		
		for(FileStatus st: status)
		{
			Path temp = st.getPath();
			FSDataInputStream in = local.open(temp);
			IOUtils.copyBytes(in, out, 4096, false);	//讀取in流中的內容放入out
			in.close();	//完成后，關閉當前文件輸入流
		}
		out.close();
	}
	public static void main(String [] args) throws IOException
	{
		String l = "/home/kqiao/hadoop/MyHadoopCodes/putmergeFiles";
		String f = "hdfs://ubuntu:9000/user/kqiao/test/PutMergeTest";
		putMergeFunc(l,f);
	}
}

3.將小文件打包成SequenceFile的MapReduce任務

來自：《hadoop權威指南》

實現將整個文件作為一條記錄處理的InputFormat:

public class WholeFileInputFormat
    extends FileInputFormat<NullWritable, BytesWritable> {
  
  @Override
  protected boolean isSplitable(JobContext context, Path file) {
    return false;
  }

  @Override
  public RecordReader<NullWritable, BytesWritable> createRecordReader(
      InputSplit split, TaskAttemptContext context) throws IOException,
      InterruptedException {
    WholeFileRecordReader reader = new WholeFileRecordReader();
    reader.initialize(split, context);
    return reader;
  }
}

實現上面類中使用的定制的RecordReader：

/實現一個定制的RecordReader，這六個方法均為繼承的RecordReader要求的虛函數。
//實現的RecordReader，為自定義的InputFormat服務
public class WholeFileRecordReader extends RecordReader<NullWritable, BytesWritable>{

	private FileSplit fileSplit;
	private Configuration conf;
	private BytesWritable value = new BytesWritable();
	private boolean processed = false;
	@Override
	public void close() throws IOException {
		// do nothing
	}

	@Override
	public NullWritable getCurrentKey() throws IOException,
			InterruptedException {
		return NullWritable.get();
	}

	@Override
	public BytesWritable getCurrentValue() throws IOException,
			InterruptedException {
		return value;
	}

	@Override
	public float getProgress() throws IOException, InterruptedException {
		return processed? 1.0f : 0.0f;
	}

	@Override
	public void initialize(InputSplit split, TaskAttemptContext context)
			throws IOException, InterruptedException {
		this.fileSplit = (FileSplit) split;
		this.conf = context.getConfiguration();
	}

	//process表示記錄是否已經被處理過
	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
	    if (!processed) {
	        byte[] contents = new byte[(int) fileSplit.getLength()];
	        Path file = fileSplit.getPath();
	        FileSystem fs = file.getFileSystem(conf);
	        FSDataInputStream in = null;
	        try {
	          in = fs.open(file);
	          	              //將file文件中 的內容放入contents數組中。使用了IOUtils實用類的readFully方法，將in流中得內容放入
	          //contents字節數組中。
	          IOUtils.readFully(in, contents, 0, contents.length);
	          //BytesWritable是一個可用做key或value的字節序列，而ByteWritable是單個字節。
	          					//將value的內容設置為contents的值
	          value.set(contents, 0, contents.length);
	        } finally {
	          IOUtils.closeStream(in);
	        }
	        processed = true;
	        return true;
	      }
	      return false;
	}
}

將小文件打包成SequenceFile：

public class SmallFilesToSequenceFileConverter extends Configured implements Tool{

	//靜態內部類，作為mapper
	static class SequenceFileMapper extends Mapper<NullWritable, BytesWritable, Text, BytesWritable>
	{
		private Text filenameKey;
		
		//setup在task開始前調用，這里主要是初始化filenamekey
		@Override
		protected void setup(Context context)
		{
			InputSplit split = context.getInputSplit();
			Path path = ((FileSplit) split).getPath();
			filenameKey = new Text(path.toString());
		}
		@Override
		public void map(NullWritable key, BytesWritable value, Context context)
				throws IOException, InterruptedException{
			context.write(filenameKey, value);
		}
	}

	@Override
	public int run(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = new Job(conf);
		job.setJobName("SmallFilesToSequenceFileConverter");
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		//再次理解此處設置的輸入輸出格式。。。它表示的是一種對文件划分，索引的方法
		job.setInputFormatClass(WholeFileInputFormat.class);
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		
		//此處的設置是最終輸出的key/value，一定要注意！
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(BytesWritable.class);
		
		job.setMapperClass(SequenceFileMapper.class);
		
		return job.waitForCompletion(true) ? 0 : 1;
	}
	
	public static void main(String [] args) throws Exception
	{
		int exitCode = ToolRunner.run(new SmallFilesToSequenceFileConverter(), args);
		System.exit(exitCode);
	}
}