归并算法在大文件处理中的使用

本文描述了一下归并算法在大文件处理中的使用.

应用场景:

1.单个文件,大小>机器内存,对文件数据进行排序(顺序,小->大)

2.单个文件,大小>机器内存,对文件数据进行去重

简单描述一下大文件排序的思路

1.文件拆分

2.拆分后的小文件分别排序,为之后的归并排序做准备

3.归并排序,这里是核心.首先,因为小文件已经排好序了,那么接下来要做的就是将有序的小文件进行合并,生成一个有序的结果文件.大概流程如下:

设置所有小文件从第一行开始读取,一次又一次的循环,循环里做的事很简单,每次循环,读取所有小文件的一行数据(如何决定读取第几行?请看下面的描述),将这些数据存入有序的list中,然后在list中取最小值,并记录索引,将最小值写入结果文件,同时标记对应的小文件,那么下一次循环时,该小文件则读取第2行,其他小文件依然读取第1行.很显然,就是每次循环,找到当前一轮读取的数据中的最小值,写入结果文件,同时将对应的小文件读取索引+1.直到所有小文件都读完最后一行数据,那么归并操作也就结束了.这里要说明的是,由于小文件已经是有序的,所以只需要在每次循环中找到当前所读数据中最小的值即可保证结果文件的有序.

关于去重,如果完全理解了排序的思路,那么对一个排好序的文件去重就相对简单了.

以上只是简单描述归并算法的使用,并未关注性能.所以接下来的实现,只是基于测试的目的,亲测后,性能是很差滴,筒子们,还是转动大脑,各显神通吧.

最后,附上 大文件排序 的代码

 OperateFile.java

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.util.Collections;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;


/**
 * 文件处理
 */

public class OperateFile {
	public static Integer UNITCOUNT = 100000;//单元数据条数
	private static Integer RANDOMSCOPE = 10000;//随机取数范围
	public static String sortpath="/tmp/sort";
	public static String splitpath="/tmp/sort/split";
	private static Random random = new Random();//
	/**
	 * 创建文件
	 * @param count
	 * @param path
	 * @throws Exception
	 */
	public void createFile(Long count,String path) throws Exception{
		BufferedWriter writer = new BufferedWriter(new FileWriter(path,false));
		Long i = 0l;
		while(i<count){
			StringBuffer tmp = new StringBuffer();
			//取2个随机数进行append组合成一个随机long值
			tmp.append(random.nextInt(RANDOMSCOPE)).append(random.nextInt(RANDOMSCOPE));
			writer.write(tmp.toString());
			writer.write("\n");
			i++;
		}
		if(writer != null){
			writer.flush();
			writer.close();
		}
	}
	/**
	 * 分割文件
	 * @param path
	 * @throws Exception
	 */
	public void splitFile(String path) throws Exception{  
		File file = new File(splitpath);
		File[] files = file.listFiles();
		for (int i = 0, n = files.length; i < n; i++) {
			files[i].delete();
		}
		BufferedReader reader = new BufferedReader(new FileReader(path));
		String tmp = null;
		StringBuffer content = new StringBuffer();
		int i = 0;
		int sum = 0;
		List<Integer> list = new LinkedList<Integer>();
		while((tmp=reader.readLine()) != null){
			if("".equals(tmp)) continue;
			list.add(Integer.parseInt(tmp));
			//如果等于或超出单元条数,则通过集合工具类排序,并生成1个新的分割文件,这里的排序,只是为后面的归并sort做准备,真实场景中肯定需要自己来处理排序
			if(list.size()>=UNITCOUNT){
				Collections.sort(list);
				for (int j = 0, n = list.size(); j < n; j++) {
					content.append(list.get(j));
					content.append("\n");
				}
				createSmallFile(content.toString(), splitpath+"/splittmp."+(i++));
				content.setLength(0);
				list.clear();
			}
			sum++;
		}
		System.out.println(sum);
		System.out.println(list.size());
		//最后还会剩下一些未达到单元条数,但又未spill到磁盘的数据,依然要生成新的分割文件
		if(list.size()>0){
			Collections.sort(list);
			for (int j = 0, n = list.size(); j < n; j++) {
				content.append(list.get(j));
				content.append("\n");
			}
			createSmallFile(content.toString(), splitpath+"/splittmp."+(i++));
			content.setLength(0);
			list.clear();
		}
		if(reader != null){
			reader.close();
		}
	}
	/**
	 * 生成分割文件
	 * @param content
	 * @param path
	 * @throws Exception
	 */
	private void createSmallFile(String content, String path) throws Exception{
		BufferedWriter writer = new BufferedWriter(new FileWriter(path,false));
		writer.write(content);
		if(writer != null){
			writer.flush();
			writer.close();
		}
	}


	public static void main(String[] args) {
		OperateFile operateFile = new OperateFile();
		try {
			long start = new Date().getTime();
			operateFile.createFile(10000000l, sortpath+"/bigfile.txt");
			long end = new Date().getTime();
			System.out.println("createFile:"+(end-start)/1000+"秒");
			start = end;
			operateFile.splitFile(sortpath+"/bigfile.txt");
			end = new Date().getTime();
			System.out.println("splitFile:"+(end-start)/1000+"秒");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

 BigFileSort.java

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.FilenameFilter;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

public class BigFileSort {
	private static Integer WRITEUNITCOUNT = 1;//100M
	private static Integer READUNITCOUNT = 1000;//100M
	private StringBuffer finish = new StringBuffer();
	private BufferedWriter writer = null;
	public static Map<String, List<Long>> splitMap = null;//split已读数据的缓存
	public static Map<String, BufferedReader> readers = null;
	/**
	 * 归并排序
	 * @param files
	 * @param target
	 * @throws Exception
	 */
	public void sortBatch(String[] files, String target) throws Exception{
		int fn = files.length;
		//存放每个split文件读取的当前行数下标
		HashMap<String,Integer> indexMap = new LinkedHashMap<String, Integer>();
		for (int i = 0; i < fn; i++) {
			indexMap.put(files[i], 0);//初始从0开始读取
		}
		while(true){
			//存放当前循环每个split文件参与排序的数据
			List<Long> list = new LinkedList<Long>();
			//存放当前循环需要进行归并操作的文件
			List<String> indexStore = new LinkedList<String>();
			int cnt = 0;
			for(String file : indexMap.keySet()){
				if(indexMap.get(file)!=null){//如果当前split文件的处理索引为null,说明该split文件的数据已全部归并完成
					//根据文件地址和已读行数来获得当前所读行的值
					Long val = readContent(file, indexMap.get(file));
					//此处的约定是,如果为null,则认为已经超出split文件行数的索引范围,视为当前文件的读已经over了,那下一轮,不再被进行归并了,因为它已经被归并完了.
					//反复几轮以后,所有split文件都会被归并,整体归并也就over了
					if(val == null){//如果返回值为null,则被认为是该split文件的数据已全部归并完成
						indexMap.put(file, null);
						cnt++ ;
					}else{
						indexStore.add(file);
						list.add(val);
					}
				}else{
					cnt++ ;
				}
			}
			if(cnt == fn) break;
			int tmpIndex = 0;
			int n = list.size();
			for (int i = 0; i < n; i++) {
				//进行比较,设置最小值的索引
				if(list.get(tmpIndex)>list.get(i)) tmpIndex=i;
			}
			try {
				//根据最小值对应的索引,获得其所在的split文件,并追加写入结果文件
				write(list.get(tmpIndex).toString()+"\n", target, finish.toString().split("\n").length>=WRITEUNITCOUNT?true:false);
			} catch (Exception e) {
				e.printStackTrace();
			}
			//更新最小值所在split文件的已读取行数
			indexMap.put(indexStore.get(tmpIndex), indexMap.get(indexStore.get(tmpIndex))+1);
		}
		if(writer != null){
			writer.flush();
			writer.close();
		}
	}
	
	/**
	 * 读取一行数据
	 * @param path
	 * @param index
	 * @return
	 * @throws Exception
	 */
	public Long readContent(String path,int index) throws Exception{
		List<Long> list = splitMap.get(path);//获得split文件读缓存
		int cnt = index/READUNITCOUNT;//整除取倍数,这里需要说明的是,为了减少频繁打开split文件所消耗的性能,此处通过临时缓存的方式来保存一次性读取的数据(缓存大小可以根据split文件数和mem大小来设置)
		if(index%READUNITCOUNT==0){//当前索引正好为整除倍数,则清空临时缓存,加载下一批数据
			BufferedReader reader = new BufferedReader(new FileReader(path));
			list.clear();
			String tmp = null;
			int i = 0;
			while((tmp=reader.readLine()) != null){
				if(i/READUNITCOUNT>cnt){//如果超出当前批次需要加载的数据,则break
					break;
				}
				if(i/READUNITCOUNT==cnt)//如果在当前批次需要加载的数据范围内,则添加至临时缓存
					list.add(Long.parseLong(tmp));
				i++;
			}
			if(reader != null) reader.close();
		}
		Long res = null;
		try {
			if(list.size()>0)//如果临时缓存中,没有数据,则返回null
				res = list.get(index%READUNITCOUNT);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return res;
	}
	
	public void write(String count,String path,boolean isWrite) throws Exception{
		finish.append(count);//append到buffer
		if(isWrite){//将buffer区域push到文件
			if(writer==null) writer = new BufferedWriter(new FileWriter(path,true));
			writer.write(finish.toString());
			finish.setLength(0);
		}
	}
	
	public static void main(String[] args) {
		BigFileSort fileSort = new BigFileSort();
		try {
			File parent = new File(OperateFile.splitpath);
			String[] files = parent.list(new FilenameFilter() {
				public boolean accept(File dir, String name) {
					if(name.indexOf("splittmp") != -1)
						return true;
					return false;
				}
			});
			splitMap = new HashMap<String, List<Long>>();
			for (int i = 0; i < files.length; i++) {
				files[i] = OperateFile.splitpath+File.separator+files[i];
				splitMap.put(files[i], new ArrayList<Long>());
			}
			File f = new File(OperateFile.sortpath+"/sort.res");
			f.delete();
			long start = new Date().getTime();
			fileSort.sortBatch(files, OperateFile.sortpath+"/sort.res");
			long end = new Date().getTime();
			System.out.println((end-start)/1000+"秒");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}