5亿整数的大文件，怎么排？

时间：2015-06-22 14:49:51 阅读：188 评论：0 收藏：0 [点我收藏+]

标签：

题目和背景可以参看这里：http://weibo.com/p/1001603856172376577500 和 http://blog.jobbole.com/87600/

这里不妨明确下题目：给定一个大文件，内含5亿个整数，每个整数都属于1-9999999之间。请设计方案，对这些元素进行排序，并将排序结果写成文件输出。可用内存不超过2G。

注意到元素一共有5亿个，因此，文件中肯定存在重复元素，也就是说有些元素不止出现一次，因此基于朴素的bitmap并不能胜任此题。

由于lau叔提供的数据集下载太麻烦，因此自己手动生成了一个。

这里提供两个方案：

方案I：由于每个元素都不超过9999999，因此，可以考虑使用CountingSort（计数排序），统计每个元素出现的次数。如果文件中1出现三次，2出现三次，5出现两次，那么最后的结果就是11122255。对CountingSort的进一步介绍，可以参考《算法导论》

#include<iostream>
using namespace std;
const int MAX = 9999999;
int C[MAX + 1] = { 0 };
void countingSort()
{
    char filename_in[] = "E:\\in.txt";
    FILE *fp_in = fopen(filename_in, "r");
    char content[10];
    while (!feof(fp_in))
    {
        fgets(content,10, fp_in);
        int element = atoi(content);
        ++C[element]; //counting...
    }
    fclose(fp_in);
    //output
    char filename_out[] = "E:\\out.txt";
    FILE *fp_out = fopen(filename_out, "w");
    for (int i = 1; i <= MAX;i++)
    {
        for (int j = 1; j <= C[MAX]; j++)
        {
            fprintf(fp_out, "%d\n", i);
        }
    }
    fclose(fp_out);
}
int main()
{
    countingSort();
    system("pause");
    return 0;
}

方案II

我们可以使用k-路归并策略，也就是原文提到的外部排序。我们先按行读入文件，分批排序，输出k个临时的小文件，其中每个小文件都是有序的。这样，我们得到k个有序的小文件，问题转换为对这k个有序小文件的合并。

说明：

1，原文评论列表中，网友iduanyingjie 的评论 “ 外部排序的第2部分【合】的时候，没必要每次去取一个最小值。因为文件1,文件2,文件3每个文件中拿出的最小值，肯定是这三个文件中的最小的3个值了，将这三个值进行排序（1,2,3），直接写入大文件。第二回合中就直接取出的是（4,5,6） ”并不正确。考虑如下的三个有序小文件：

20 50 60

40 45 70

70 85 90

首选取出20、40、70，排序比较后输出最小的元素20，此时正确的做法是从第一个文件（也就是最小元素20所在的那个文件）读出50，然后从50、40、70中选出最小元素，也就是50，输出。接着从50所在的文件也就是第一个文件读出60，从60、40、70选出最小元素也就是40输出。以此类推。此时，并不能按照网友iduanyingjie 所说的简单的排序输出、排序输出。

因此，正确的做法是，读入k个元素，输出最小元素，再从最小元素所在的文件读下一个元素，继续选择最小元素输出；直到k个有序文件中的所有元素都处理完毕为止。

2，由于选择最小值是一个很关键的操作，因此可以使用包含k个元素的最小堆来高效的完成。整个过程就转换为：输出最小、替换堆顶、重建堆；输出最小、替换堆顶、重建堆......

3，当某个小文件的元素都处理完毕后，此时我们采取的策略是移动堆底部的元素到堆顶，并将堆大小减去1。这样，当堆大小为0的时候，说明所有的文件中的所有元素都处理完毕了。为了知道最小元素所在的文件，我们需要一个变量来记录文件号。

#include<iostream>
using namespace std;
const int K = 500;
struct Node 
{
    int value;
    int file_id;
};
class Heap
{
public:
    Heap(int capacity)
    {
        this->capacity = capacity;
        this->size = 0;
        p = new Node[this->capacity + 1];
    }
    void buildMinHeap() 
    {
        for (int i = size / 2; i >= 1; i--)
        {
            minHeapify(i);
        }            
    }
    void insertNode(Node t)
    {
        size++;
        p[size].value = t.value;
        p[size].file_id = t.file_id;
    }
    Node getMin()
    {
        return p[1];
    }
    int getHeapSize()
    {
        return size;
    }
    void minHeapify(int i) 
    {
        int left = 2 * i;
        int right = 2 * i + 1;
        int min = i;
        if (left <= size && p[left].value < p[i].value)
        {
            min = left;
        }            
        if (right <= size && p[right].value < p[min].value)
        {
            min = right;
        }            
        if (min != i) 
        {
            Node tmp = p[i];
            p[i] = p[min];
            p[min] = tmp;
            minHeapify(min);
        }
    }
    void replaceRootNodeByLastNode()
    {
        p[1] = p[size];
        size--;
    }
    void replaceRootNodeByNextElement(Node n)
    {
        p[1] =n;
    }
    ~Heap()
    {
        delete[]p;
    }
private:
    Node *p;
    int size;
    int capacity;
};
Node getOneElement(FILE* & fp,int file_id)
{
    char content[10];
    Node ret;
    if(!feof(fp))
    {
        fgets(content, 10, fp);
        int element = atoi(content);
        ret.file_id = file_id;
        ret.value = element;
    }
    else
    {
        ret.value = -1;
    }
    return ret;
}
void writeResult(FILE* & fp, int element)
{
    fprintf(fp, "%d\n", element);
}
void kWayMergeViaHeap(FILE** fp)
{
    Heap *pHeap = new Heap(K);
    for (int i = 0; i <K; i++)
    {
        Node t = getOneElement(fp[i], i);
        pHeap->insertNode(t);
    }
    pHeap->buildMinHeap();
    char filename_output[] = "E:\\outT.txt";
    FILE *fp_out = fopen(filename_output, "w");
    while (pHeap->getHeapSize()>0) 
    {
        Node minNode = pHeap->getMin();
        writeResult(fp_out, minNode.value);
        Node t = getOneElement(fp[minNode.file_id], minNode.file_id);
        if (t.value==-1) //there are no elements in fp[minNode.file_id] 
        {
            pHeap->replaceRootNodeByLastNode();
        }
        else 
        {
            pHeap->replaceRootNodeByNextElement(t);
        }
        pHeap->minHeapify(1);
    }
    //
    for (int i = 0; i <K; i++)
    {
        fclose(fp[i]);
    }
    fclose(fp_out);
    delete pHeap;
}

1，堆的实现大学时候写的，参照《算法导论》完成，尽管进一步的加速和优化是可能的。例如，minHeapify函数的末尾是一个尾递归，可以通过循环来代替。当然啦，开启编译器优化之后编译器可能会自动完成，因此2*i这类语句就真没必要用移位操作了。再例如，getOneElement中的接口设计，返回Node意味着需要构造函数开销，当然设计为只返回int更佳，但是后面就稍微麻烦了点。