大数据去除重复--实战（二）

           关于上一篇数据去重复的问题，在结尾的时候提到，另一种思路：在url-->hashCode 根据范围写入文件的时候，不用迭代二分法，采用平均算法，也就是说根据url的大概行数，设置一个单位区间，循环遍历行的时候，根据hashCode 值，放入不同的空间，然后再放入内存去除重复，写入汇总文件。

          去个例子，我文件数据2G，1.5亿行，自己设定一个区间值1000W。 也就是说0~1000W、1000W~2000W 。。。为一个区间，每行都会通过计算放入不同的区间，这样在数据均匀的情况下，分布式很均匀的，如果某一个区间值过多，超过内存限制，继续切割，这样会大大减少第一种算法带来的文件数据的重复读取的效率。

          看代码，这里生产文件的代码参考前一篇文章。

          

        

package com.files;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.Reader;
import java.io.Writer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

public class PartFile2 {
	// 内存监控
	final static Runtime currRuntime = Runtime.getRuntime ();
	// 最小的空闲空间，额，可以用来 智能控制，- -考虑到GC ，暂时没用
	final static long MEMERY_LIMIT = 1024 * 1024 * 3;
	// 内存限制，我内存最多容纳的文件大小
	static final long FILE_LIMIT_SIZE = 1024*1024*1;
	// 文件写入缓冲区 ，我默认
	static final int CACHE_SIZE = 1024;
	// 默认文件后缀
	static final String FILE_SUFFIX = ".txt";
	// 临时分割的文件目录，可以删除~。~
	static final String FILE_PREFIX = "test/";
	// 汇总的文件名
	static final String REQUST_FILE_NAME = "resultFile.txt";

	
	// 存放小文件的，驱除重复数据
	static Map<String,String> fileLinesMap = new HashMap<String,String>(10000);
	// 按hashCode 分割的单位
	static int HASH_UNIT = 1000*1000*10;
	// 收缩率，HASH_UNIT 在一定范围内，数据过大，用这个调节之后继续分割
	static final int shrinkage_factor = 1000;
	// 存放分割d
	static Map<File,BufferedWriter> map = new HashMap<File, BufferedWriter>();
	
	// 存放大文件 引用，以及分割位置
	static List<ChildFile> bigChildFiles = new ArrayList<ChildFile>();
	
	static final String origFileName = "xxx.txt";

	public static void main(String[] args) {
		long begin = System.currentTimeMillis();
		new PartFile2().partFile(new File(origFileName));
		long result = System.currentTimeMillis()-begin;
		System.out.println("除去重复时间为："+result +" 毫秒");
		// 除去重复时间为：931594 毫秒
	}
     
	
	// 按hashCode 范围分割
	public  void partFile(File origFile) {
		String line = null;
		BufferedReader reader = null;
		int hashCode;
		try {
			reader = new BufferedReader(new FileReader(origFile));
			while ((line = reader.readLine()) != null) {
				hashCode = line.hashCode(); 
				File file = getFileByHashCode(hashCode);
				writeToFile(getFileWriterByHashCode(hashCode,file), line);
			}
			// 将集合对象可以内存排序的全部写入汇总文件，大于内存的继续分割
			Iterator<Entry<File, BufferedWriter>> it = map.entrySet().iterator();
			while(it.hasNext()){
				Entry<File, BufferedWriter> entry = it.next();
				File file = entry.getKey();
				BufferedWriter writer = entry.getValue();
				if(isSurpassFileSize(file)){
					// 这里本想用再次细化空间，进行分割，发现当大量重复的元素的时候会溢出
					// 极端的情况下，全是重复元素，即使以前的方法也不能执行，需要另外控制
					// HASH_UNIT = HASH_UNIT/shrinkage_factor;
					// partFile(file);
					
					// 如果超出，继续分割
					String[] number = file.getName().split("_");
					long maxNum = Long.parseLong(number[0])*HASH_UNIT;
					long minNum = Long.parseLong(number[1].substring(0, number[1].indexOf(FILE_SUFFIX)))*HASH_UNIT;
					partFile(file,maxNum,minNum);
				}else{
					writer.flush();
					orderAndWriteToFiles(file);
					// 关闭刷新流
					closeWriter(writer);
					file.delete();
				}
			}
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}finally{
			closeReader(reader);
			shotDown();
		}
	}
	
	// 清除临时文件
	public void shotDown(){
		Iterator<File> it = map.keySet().iterator();
		while(it.hasNext()){
			File file = it.next();
			file.delete();
		}
	}
	
	// 按hashCode 范围分割
	public  void partFile(File origFile,long maxNum,long minNum) {
		String line = null;
		long hashCode = 0;
		long max_left_hashCode = 0;
		long min_left_hashCode = 0;
		long max_right_hashCode = 0;
		long min_right_hashCode = 0;
		BufferedWriter rightWriter = null;
		BufferedWriter leftWriter = null;
		BufferedReader reader = null;
		try {
			reader = new BufferedReader(new FileReader(origFile));
			long midNum = (maxNum+minNum)/2;
			// 以文件hashCode 范围作为子文件名
			File leftFile = new File(FILE_PREFIX+minNum+"_"+midNum+FILE_SUFFIX);
			File rightFile = new File(FILE_PREFIX+midNum +"_"+maxNum+FILE_SUFFIX);
			
			leftWriter = new BufferedWriter(new FileWriter(leftFile),CACHE_SIZE);
			rightWriter = new BufferedWriter(new FileWriter(rightFile),CACHE_SIZE);
			
			ChildFile leftChild = new ChildFile(leftFile);
			ChildFile rightChild = new ChildFile(rightFile);
			
			// hashCode 的范围作为分割线
			while ((line = reader.readLine()) != null) {
				hashCode = line.hashCode();
					if (hashCode > midNum) {
						if(max_right_hashCode < hashCode || max_right_hashCode == 0){
							max_right_hashCode = hashCode;
						}else if(min_right_hashCode > hashCode || min_right_hashCode == 0){
							min_right_hashCode = hashCode;
						}
						// 按行写入缓存
						writeToFile(rightWriter, line);
				}else {
						if(max_left_hashCode < hashCode || max_left_hashCode == 0){
							max_left_hashCode = hashCode;
						}else if(min_left_hashCode > hashCode || min_left_hashCode == 0){
							min_left_hashCode = hashCode;
						}
						writeToFile(leftWriter, line);
				}
			}
			// 保存子文件信息
			leftChild.setHashCode(min_left_hashCode, max_left_hashCode);
			rightChild.setHashCode(min_right_hashCode, max_right_hashCode);
			
			closeWriter(rightWriter);
			closeWriter(leftWriter);
			closeReader(reader);

			
			// 删除原始文件，保留最原始的文件
			
			if(!origFile.getName().equals(origFileName)){
				origFile.delete();
			}
			
			// 分析子文件信息，是否写入或者迭代
			analyseChildFile(rightChild, leftChild);

		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	
	// 分析子文件信息
	public void analyseChildFile(ChildFile rightChild,ChildFile leftChild){
		// 将分割后 还是大于内存的文件保存  继续分割
		File rightFile = rightChild.getChildFile();
		if(isSurpassFileSize(rightFile)){
			bigChildFiles.add(rightChild);
		}else if(rightFile.length()>0){
			orderAndWriteToFiles(rightFile);
		}
		
		File leftFile = leftChild.getChildFile();
		if(isSurpassFileSize(leftFile)){
			bigChildFiles.add(leftChild);
		}else if(leftFile.length()>0){
			orderAndWriteToFiles(leftFile);
		}
		
		// 未超出直接内存排序，写入文件，超出继续分割，从末尾开始，不易栈深度溢出
		if(bigChildFiles.size() > 0 ){
			ChildFile e  = bigChildFiles.get(bigChildFiles.size()-1);
			bigChildFiles.remove(e);
			// 迭代分割
			partFile(e.getChildFile(), e.getMaxHashCode(), e.getMinHashCode());
		}
	}
	
	
	// 根据hashCode 值，计算存放的位置
	public File getFileByHashCode(int hashCode){
		int i = hashCode/HASH_UNIT;
		return new File(FILE_PREFIX+i+"_"+i+1+FILE_SUFFIX);
	}
	
	// 获得缓存流，这里缓存大小不能太大
	public BufferedWriter getFileWriterByHashCode(int hashCode,File file) throws IOException{
		BufferedWriter writer = null;
		writer = map.get(file);
		if(writer == null){
			writer = new BufferedWriter(new FileWriter(file,true),CACHE_SIZE);
			map.put(file, writer);
		}
		return writer;
	}
	

	// 将小文件读到内存排序除重复
	public  void orderAndWriteToFiles(File file){
		BufferedReader reader = null;
		String line = null;
		BufferedWriter totalWriter = null;
		StringBuilder sb = new StringBuilder(1000000);
		try {
			totalWriter = new BufferedWriter(new FileWriter(REQUST_FILE_NAME,true),CACHE_SIZE);
			reader = new BufferedReader(new FileReader(file));
			while((line = reader.readLine()) != null){
				if(!fileLinesMap.containsKey(line)){
					fileLinesMap.put(line, null);
					//sb.append(line+"\r\n");
					totalWriter.write(line+"\r\n");
				}
			}
			//totalWriter.write(sb.toString());
			fileLinesMap.clear();
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}finally{
			closeReader(reader);
			closeWriter(totalWriter);
			file.delete();
		}
	}
	

	
	// 判断该文件是否超过 内存限制
	public  boolean isSurpassFileSize(File file){
		if(file.length() == 0){
			System.out.println(file.delete());;
			return false;
		}
		return FILE_LIMIT_SIZE <  file.length();
	}
	

	// 将数据写入文件
	public  void writeToFile(BufferedWriter writer, String writeInfo) {
		try {
			writer.write(writeInfo+"\r\n");
		} catch (IOException e) {
			e.printStackTrace();
		}finally{
		}
	}

	// 关闭流
	public  void closeReader(Reader reader) {
		if (reader != null) {
			try {
				reader.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	// 关闭流
	public  void closeWriter(Writer writer) {
		if (writer != null) {
			try {
				writer.flush();
				writer.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	
	// 内部类，记录子文件信息
	class ChildFile{
		// 文件 和 内容 hash 分布
		File childFile;
		long maxHashCode;
		long minHashCode;
		
		public ChildFile(File childFile){
			this.childFile = childFile;
		}
		
		public ChildFile(File childFile, long maxHashCode, long minHashCode) {
			super();
			this.childFile = childFile;
			this.maxHashCode = maxHashCode;
			this.minHashCode = minHashCode;
		}
		
		public File getChildFile() {
			return childFile;
		}
		public void setChildFile(File childFile) {
			this.childFile = childFile;
		}
		public long getMaxHashCode() {
			return maxHashCode;
		}
		public void setMaxHashCode(long maxHashCode) {
			this.maxHashCode = maxHashCode;
		}
		public long getMinHashCode() {
			return minHashCode;
		}
		public void setMinHashCode(long minHashCode) {
			this.minHashCode = minHashCode;
		}
		
		public void setHashCode(long minHashCode,long maxHashCode){
			this.setMaxHashCode(maxHashCode);
			this.setMinHashCode(minHashCode);
		}
		
	}

}

 

在数据分布均匀的均匀，不会有大量重复数据的情况下，比如用上一次的2G 文件测试，能节约5分钟左右的时间，如果全是重复数据会溢出失败，不太稳定，需要对数据做一定量的分析。

 

 

 

小结：

           1.这是对实战一方法的一些改进，在一定程度上能提高速度。小数据（100m）和第一种速度相当，在数据hashCode分布均匀，速度较快。

           2.在处理这类大数据的问题上，主要是分离法，hash分离 成内存可以容纳的文件，然后分别处理，不同的要求有不同的处理办法，但是大致方法类似。

           3.有问题，请留言，共同探讨！