外部排序 - 元素码农

发布时间: 2025-03-21 18:52

↑

# 外部排序

外部排序（External Sort）是在数据量非常大，无法一次性加载到内存中时使用的排序算法。它通常采用分块排序和多路归并的策略，将大文件分成多个小文件进行排序，然后合并这些有序文件。

## 基本原理

1. 分块：将大文件分成多个能够装入内存的小文件
2. 内部排序：对每个小文件进行内部排序
3. 多路归并：将已排序的小文件合并成一个大的有序文件

## 工作流程

```
原始大文件：
[89 45 23 76 12 99 34 67 ...]

分块：
块1：[89 45 23]  -> 排序后：[23 45 89]
块2：[76 12 99]  -> 排序后：[12 76 99]
块3：[34 67 ...]  -> 排序后：[34 67 ...]

多路归并：
[23 45 89] + [12 76 99] + [34 67 ...]
-> [12 23 34 45 67 76 89 99 ...]
```

## 代码实现

### 基本实现

```python
class ExternalSort:
    def __init__(self, chunk_size=1000):
        self.chunk_size = chunk_size
    
    def split_file(self, input_file):
        """将输入文件分割成多个小文件"""
        chunk_files = []
        chunk = []
        chunk_index = 0
        
        with open(input_file, 'r') as f:
            for line in f:
                num = int(line.strip())
                chunk.append(num)
                
                if len(chunk) >= self.chunk_size:
                    # 对当前块进行排序
                    chunk.sort()
                    
                    # 写入临时文件
                    chunk_file = f'chunk_{chunk_index}.txt'
                    with open(chunk_file, 'w') as cf:
                        for num in chunk:
                            cf.write(f"{num}\n")
                    
                    chunk_files.append(chunk_file)
                    chunk = []
                    chunk_index += 1
        
        # 处理最后一个块
        if chunk:
            chunk.sort()
            chunk_file = f'chunk_{chunk_index}.txt'
            with open(chunk_file, 'w') as cf:
                for num in chunk:
                    cf.write(f"{num}\n")
            chunk_files.append(chunk_file)
        
        return chunk_files
    
    def merge_files(self, chunk_files, output_file):
        """合并多个有序文件"""
        from heapq import heappush, heappop
        
        # 打开所有输入文件
        files = [open(f) for f in chunk_files]
        heap = []
        
        # 从每个文件读取第一个数
        for i, f in enumerate(files):
            line = f.readline()
            if line:
                num = int(line.strip())
                heappush(heap, (num, i))
        
        # 开始归并
        with open(output_file, 'w') as out:
            while heap:
                num, i = heappop(heap)
                out.write(f"{num}\n")
                
                # 从对应文件读取下一个数
                line = files[i].readline()
                if line:
                    next_num = int(line.strip())
                    heappush(heap, (next_num, i))
        
        # 关闭所有文件
        for f in files:
            f.close()
        
        # 删除临时文件
        import os
        for f in chunk_files:
            os.remove(f)
    
    def sort(self, input_file, output_file):
        """执行外部排序"""
        # 分割文件
        chunk_files = self.split_file(input_file)
        
        # 合并排序后的文件
        self.merge_files(chunk_files, output_file)

# 使用示例
sorter = ExternalSort(chunk_size=1000)
sorter.sort('input.txt', 'output.txt')
```

### 优化版本

```python
class OptimizedExternalSort:
    def __init__(self, chunk_size=1000, buffer_size=4096):
        self.chunk_size = chunk_size
        self.buffer_size = buffer_size
    
    def create_chunk(self, numbers):
        """创建并排序一个数据块"""
        numbers.sort()
        chunk_file = f'chunk_{len(self._chunk_files)}.txt'
        with open(chunk_file, 'w', buffering=self.buffer_size) as f:
            for num in numbers:
                f.write(f"{num}\n")
        self._chunk_files.append(chunk_file)
    
    def split_file(self, input_file):
        """使用缓冲读取分割文件"""
        self._chunk_files = []
        current_chunk = []
        
        with open(input_file, 'r', buffering=self.buffer_size) as f:
            for line in f:
                current_chunk.append(int(line.strip()))
                
                if len(current_chunk) >= self.chunk_size:
                    self.create_chunk(current_chunk)
                    current_chunk = []
        
        if current_chunk:
            self.create_chunk(current_chunk)
        
        return self._chunk_files
    
    def merge_chunks(self, output_file, batch_size=10):
        """分批次合并文件"""
        while len(self._chunk_files) > 1:
            new_chunks = []
            
            for i in range(0, len(self._chunk_files), batch_size):
                batch = self._chunk_files[i:i + batch_size]
                output_chunk = f'merged_{len(new_chunks)}.txt'
                self.merge_batch(batch, output_chunk)
                new_chunks.append(output_chunk)
                
                # 删除已合并的文件
                for f in batch:
                    import os
                    os.remove(f)
            
            self._chunk_files = new_chunks
        
        # 重命名最后一个文件为输出文件
        import os
        os.rename(self._chunk_files[0], output_file)
    
    def merge_batch(self, input_files, output_file):
        """合并一批文件"""
        from heapq import heappush, heappop
        
        files = [open(f, 'r', buffering=self.buffer_size) for f in input_files]
        heap = []
        
        # 初始化堆
        for i, f in enumerate(files):
            line = f.readline()
            if line:
                heappush(heap, (int(line.strip()), i))
        
        with open(output_file, 'w', buffering=self.buffer_size) as out:
            while heap:
                num, i = heappop(heap)
                out.write(f"{num}\n")
                
                line = files[i].readline()
                if line:
                    heappush(heap, (int(line.strip()), i))
        
        for f in files:
            f.close()
    
    def sort(self, input_file, output_file):
        """执行优化后的外部排序"""
        chunk_files = self.split_file(input_file)
        self.merge_chunks(output_file)

# 使用示例
sorter = OptimizedExternalSort(chunk_size=1000, buffer_size=8192)
sorter.sort('large_input.txt', 'sorted_output.txt')
```

## 性能优化

### 内存管理

1. 合理设置块大小
```python
def calculate_chunk_size(total_memory, safety_factor=0.75):
    """根据可用内存计算合适的块大小"""
    import psutil
    available_memory = psutil.virtual_memory().available
    chunk_size = int((available_memory * safety_factor) / 8)  # 假设每个数占8字节
    return min(chunk_size, total_memory)
```

2. 使用缓冲区
```python
def create_buffered_reader(file_path, buffer_size=8192):
    """创建带缓冲的文件读取器"""
    return open(file_path, 'r', buffering=buffer_size)
```

### 多线程优化

```python
from concurrent.futures import ThreadPoolExecutor

class ParallelExternalSort:
    def __init__(self, chunk_size=1000, num_threads=4):
        self.chunk_size = chunk_size
        self.num_threads = num_threads
    
    def sort_chunk(self, chunk_data):
        """并行排序数据块"""
        chunk_data.sort()
        return chunk_data
    
    def split_and_sort(self, input_file):
        chunks = []
        current_chunk = []
        
        with ThreadPoolExecutor(max_workers=self.num_threads) as executor:
            futures = []
            
            with open(input_file, 'r') as f:
                for line in f:
                    current_chunk.append(int(line.strip()))
                    
                    if len(current_chunk) >= self.chunk_size:
                        futures.append(
                            executor.submit(self.sort_chunk, current_chunk)
                        )
                        current_chunk = []
            
            if current_chunk:
                futures.append(
                    executor.submit(self.sort_chunk, current_chunk)
                )
            
            # 获取排序后的块
            for future in futures:
                chunks.append(future.result())
        
        return chunks
```

## 应用场景

1. 大文件排序
2. 数据库索引构建
3. 日志文件处理
4. 大规模数据分析

## 实际应用示例

### 1. 日志文件排序

```python
class LogSorter:
    def __init__(self, chunk_size=1000):
        self.sorter = OptimizedExternalSort(chunk_size)
    
    def parse_log_line(self, line):
        """解析日志行，返回时间戳"""
        import datetime
        timestamp_str = line.split('[')[1].split(']')[0]
        timestamp = datetime.datetime.strptime(
            timestamp_str, "%Y-%m-%d %H:%M:%S"
        )
        return timestamp, line
    
    def sort_logs(self, input_log, output_log):
        # 创建临时文件存储时间戳和行号
        temp_file = 'temp_timestamps.txt'
        line_map = {}
        line_number = 0
        
        # 提取时间戳
        with open(input_log, 'r') as f, open(temp_file, 'w') as temp:
            for line in f:
                timestamp, _ = self.parse_log_line(line)
                unix_time = timestamp.timestamp()
                temp.write(f"{unix_time} {line_number}\n")
                line_map[line_number] = line
                line_number += 1
        
        # 排序时间戳文件
        self.sorter.sort(temp_file, 'sorted_temp.txt')
        
        # 重建排序后的日志文件
        with open('sorted_temp.txt', 'r') as temp, \
             open(output_log, 'w') as out:
            for line in temp:
                _, line_num = line.strip().split()
                out.write(line_map[int(line_num)])
        
        # 清理临时文件
        import os
        os.remove(temp_file)
        os.remove('sorted_temp.txt')

# 使用示例
log_sorter = LogSorter()
log_sorter.sort_logs('server.log', 'sorted_server.log')
```

### 2. 大文件去重

```python
class LargeFileDeduplicator:
    def __init__(self, chunk_size=1000):
        self.sorter = OptimizedExternalSort(chunk_size)
    
    def deduplicate(self, input_file, output_file):
        # 首先排序文件
        self.sorter.sort(input_file, 'sorted_temp.txt')
        
        # 读取排序后的文件并去重
        prev_line = None
        with open('sorted_temp.txt', 'r') as f, \
             open(output_file, 'w') as out:
            for line in f:
                current_line = line.strip()
                if current_line != prev_line:
                    out.write(line)
                    prev_line = current_line
        
        # 清理临时文件
        import os
        os.remove('sorted_temp.txt')

# 使用示例
deduplicator = LargeFileDeduplicator()
deduplicator.deduplicate('large_data.txt', 'unique_data.txt')
```

## 总结

外部排序是处理大规模数据的重要算法，它的主要特点是：

1. 适用于无法一次性加载到内存的大数据集
2. 通过分块和归并策略实现
3. 可以通过多种方式优化性能
4. 在实际应用中有广泛用途

在实现外部排序时，需要注意以下几点：

1. 合理设置块大小和缓