内存不足时的大文件排序算法（spark shuffle的排序算法）

1、问题场景

例如：当前磁盘2T，内存16G，文件大小500G。现在需要对500G的大文件进行排序，并存道另一个文件中。

抽象：当前有包含10000行数的文件，再只能使用长度为1000的数组的前提下，对文件中的数据进行排序，并存道另一个文件中。

2、问题分析

首先，数据的总量大于了所能使用的内存数据结构的容量，所以肯定不能完全通过内存排序。因此需要采用一种分治的思想来实现全局排序。

（1）数组总量10000，数组容量1000，所以可以分十次读取数据，每次读取1000条数据，再内存中对1000条数据进行排序。

（2）单次排完序的数据写入磁盘临时文件。总供会有10个局部有序的磁盘文件。

（3）同时打开10个磁盘文件，创建reader。每个磁盘文件都读入第一行数据，即局部最小的数据。

（4）找到10个局部最小数据中的最小值。

（5）对每个磁盘文件，判断当前最小值是否等于全局最小值。如果等于则写出，并获取下一行数据，继续比较。

（6）当所有文件都比较并输出完之后，重新计算当前10个局部最小数据中的最小值。继续比较输出，直到所有磁盘文件全部读完。

（7）此时输出的文件就是一个全局有序的文件。

3、生成测试数据

public class CreateTestData {public static void main(String[] args) throws IOException {String filePath = "C:\\Users\\yangshen\\code\\java\\stu-project\\src\\main\\resources\\input.txt";BufferedWriter writer = new BufferedWriter(new FileWriter(filePath));Random random = new Random(1);int num = 0;while (num < 10000) {writer.write(String.valueOf(random.nextInt(10000)));writer.newLine();num++;}writer.close();}
}

4、排序算法

public class SortWithMuchData {public static void main(String[] args) throws Exception {String fileIn = "C:\\Users\\yangshen\\code\\java\\stu-project\\src\\main\\resources\\input.txt";String fileOut = "C:\\Users\\yangshen\\code\\java\\stu-project\\src\\main\\resources\\output.txt";new Sort(fileIn, fileOut, 1000).run();}
}class Sort {private final String fileIn;private final String fileOut;private int[] data;private final int size;private final List<String> tempFiles;public Sort(String fileIn, String fileOut, int size) {this.fileIn = fileIn;this.fileOut = fileOut;this.data = new int[size];this.size = size;this.tempFiles = new ArrayList<>();}public void run() throws Exception {System.out.println("begin...");BufferedReader reader = new BufferedReader(new FileReader(fileIn));String line;int num = 0;while ((line = reader.readLine()) != null) {if (num < size) {data[num] = Integer.parseInt(line);num++;} else {sortAndSpill();data = new int[size];num = 0;}}if (data.length > 0) {sortAndSpill();}merge();clear();System.out.println("end...");}/*** 合并临时文件*/private void merge() throws Exception {new Merge(tempFiles, fileOut).run();}/*** 清理临时文件*/private void clear() {for (String tempFile : tempFiles) {File file = new File(tempFile);if (file.exists()) {file.delete();}}}/*** 数组排序，持久化*/private void sortAndSpill() throws IOException {Arrays.sort(data);spill();}/*** 持久化临时文件*/private void spill() throws IOException {String tempFileName = createTempFileName();BufferedWriter writer = new BufferedWriter(new FileWriter(tempFileName));for (int d : data) {writer.write(String.valueOf(d));writer.newLine();}writer.close();System.out.println("生成临时文件：" + tempFileName);tempFiles.add(tempFileName);}/*** 生成临时文件名** @return 临时文件名*/private String createTempFileName() {return "C:\\Users\\yangshen\\code\\java\\stu-project\\src\\main\\resources\\" + UUID.randomUUID().toString() + ".txt";}static class Merge {private int minValue;private final int fileNum;private final int[] tmpData;private final BufferedReader[] readers;private final String[] tempFiles;private final BufferedWriter writer;public Merge(List<String> tempFiles, String fileOut) throws IOException {this.minValue = -1;this.fileNum = tempFiles.size();this.tempFiles = tempFiles.toArray(new String[this.fileNum]);this.tmpData = new int[this.fileNum];this.readers = new BufferedReader[this.fileNum];this.writer = new BufferedWriter(new FileWriter(fileOut));init();}private void init() throws IOException {for (int i = 0; i < tempFiles.length; i++) {readers[i] = new BufferedReader(new FileReader(tempFiles[i]));tmpData[i] = Integer.parseInt(readers[i].readLine());}}public void run() throws IOException {while (true) {int ok = 0;resetMinValue();for (int i = 0; i < tmpData.length; i++) {while (tmpData[i] == minValue) {System.out.println(tmpData[i]);write(String.valueOf(tmpData[i]));fillTempData(i);}if (tmpData[i] == -1) {ok++;}}if (ok == fileNum) {break;}}writer.close();}/*** 填充临时数组** @param i 位置*/private void fillTempData(int i) throws IOException {String line;if ((line = readers[i].readLine()) != null) {tmpData[i] = Integer.parseInt(line);} else {tmpData[i] = -1;}}/*** 写出** @param item 写出内容*/private void write(String item) throws IOException {writer.write(item);writer.newLine();}/*** 更新最小值*/private void resetMinValue() {minValue = Arrays.stream(tmpData).filter(x -> x != -1).min().getAsInt();}}
}

内存不足时的大文件排序算法（spark shuffle的排序算法）相关推荐

python让词向量一直在内存里_python读取大文件踩过的坑——读取txt文件词向量
在读取https://github.com/Embedding/Chinese-Word-Vectors中的中文词向量时,选择了一个有3G多的txt文件,之前在做词向量时用的是word2vec,所以直 ...
文件内存映射mmap解决大文件快速读写问题和进程间共享内存
mmap函数主要用途有三个: 1.将一个普通文件映射到内存中,通常在需要对文件进行频繁读写时使用,这样用内存读写取代I/O读写,以获得较高的性能: 2.将特殊文件进行匿名内存映射,可以为关联进程提供共 ...
digital ocean 内存不足时增加swap文件的方法
买了比较低配的digitalocean 云主机,在执行composer update的时候出现内存不足的问题,但是内存大小已经固定了,除非加钱,还有别的方法吗? 有,增加swap分区,这样就可以弥补内 ...
数据算法——Spark二次排序
1.Scala实现: /*** 二次排序:超过2列(特征)* 对比MR天气案例,自定义一个key(包含读到的一行数字),对key进行内部比较.*/ object SecondSort {def mai ...
大文件多路归并排序
1 题目这一种题目的描述,大概有以后两种: 题目1:一个大文件在一台服务器上存不下,需要存放在多台服务器上,将这个大文件的内容进行排序. 题目2:一个大文件100G,存储在磁盘上,现在需要对这个文件 ...
海量数据处理面试题/大文件问题
首先引用原博主的一句话,感触颇深面试时,70~80%的人不是倒在这两方面,而是倒在基础之上(诸如语言,数据库,操作系统,网络协议等等),所以,无论任何时候,基础最重要,没了基础,便什么都不是. 不要 ...
多线程大文件断点续传和流媒体的处理方法
2019独角兽企业重金招聘Python工程师标准>>> 在使用Squid做反向代理的CDN节点时.多线程大文件断点续传和流媒体的处理是怎么样啦.前些日子花了点时间研究了一下. 在Sq ...
java 处理大文件
目的: 前几天在开发过程中遇到一个需求: 读取一个大约5G的csv文件内容,将其转化为对象然后存储到redis中, 想着直接开大内存直接load 进入到内存中就行了,结果可想而知,5G的文件 ,Xmx ...
前后端配合实现大文件断点续传（前端逻辑）
断点续传作用:当上传文件时,大文件上传时耗时过长,如果遇到网络卡顿.断网等情况,再重新开始上传的体验感非常不好.前端优化分片上传文件,上传时把大文件分成很多个小文件,等到网络状况恢复了之后,即从之前上 ...