선택 심화: 런·다중 병합·외부 정렬

지금까지의 정렬은 모두 정렬할 값 전체가 메모리, 곧 RAM에 들어 있다고 가정했습니다. 하지만 서버의 기록 파일이나 대규모 측정 자료처럼 정렬할 자료가 메모리보다 훨씬 큰 경우도 있습니다. 이때는 자료를 조금씩 메모리로 읽어 처리하고 결과를 다시 파일로 써야 합니다. 이렇게 자료가 주로 메모리에 있는 정렬을 내부 정렬(internal sorting), 파일 같은 외부 저장 장치에 있는 자료를 제한된 메모리로 정렬하는 것을 외부 정렬(external sorting)이라고 합니다. 이 글에서는 병합 정렬과 힙을 바탕으로 외부 정렬을 작은 파일로 직접 실행해 보고, 메모리와 파일의 제약이 알고리즘을 어떻게 바꾸는지 살펴봅니다.

두 단계로 정렬한다

외부 정렬은 두 단계로 이루어집니다. 첫 단계에서는 입력 파일을 메모리에 들어가는 크기만큼씩 읽어, 그 묶음을 메모리 안에서 정렬한 뒤 임시 파일로 씁니다. 한 번에 읽는 묶음을 청크(chunk)라 하고, 이렇게 만든 정렬된 조각을 런(run)이라고 합니다. 두 번째 단계에서는 런 여러 개를 하나의 정렬된 출력으로 합칩니다. 병합 정렬이 정렬된 두 리스트를 합쳤던 것처럼, 여기서는 정렬된 런 k개를 한꺼번에 합치는 다중 병합(k-way merge)을 합니다.

청크 크기가 3이고 입력이 8, 1, 6, 3, 9, 2, 7이라면, 첫 단계에서 세 개씩 읽어 정렬한 런 [1, 6, 8], [2, 3, 9], [7]이 만들어집니다. 마지막 청크는 값이 하나만 남았지만 그대로 런이 됩니다.

맨 앞 값만 비교해 합친다

런이 셋이면 세 런의 맨 앞 값 가운데 가장 작은 것이 남은 모든 값 가운데 가장 작은 값입니다. 각 런은 정렬되어 있어서 맨 앞 값이 그 런의 최솟값이기 때문입니다. 그 값을 출력하고 그 값이 나온 런에서만 다음 값을 읽어 새 맨 앞 값으로 삼는 일을 되풀이하면, 정렬된 출력이 만들어집니다. 다 읽은 런은 소진되었다고 하며, 더는 비교하지 않습니다.

어느 순간에도 메모리에 있어야 하는 값은 런마다 맨 앞 값 하나씩뿐입니다. 런이 아무리 길어도 전체 값을 한꺼번에 읽을 필요가 없는 것이 이 방법의 핵심입니다. 맨 앞 값 k개 가운데 가장 작은 것을 고르는 일은 우선순위 큐가 잘하는 일이라서, 최소 힙에 맨 앞 값을 넣어 두면 값 하나를 출력할 때마다 O(log k)에 다음 최솟값을 고를 수 있습니다.

파일로 실행해 보기

Python 표준 라이브러리의 heapq.merge는 이미 정렬된 여러 반복 가능한 객체를 받아, 힙으로 맨 앞 값들을 비교하며 합친 값을 하나씩 내놓습니다. 아래 코드는 이것으로 외부 정렬을 구현합니다. 실제로 큰 파일을 쓰지는 않고, 임시 폴더에 숫자 일곱 개짜리 입력 파일을 만들어 실행합니다.

파일을 다루는 데 필요한 도구 몇 가지를 먼저 정리하겠습니다. Path는 파일이나 폴더의 경로를 나타내는 객체이고, path.open(encoding="utf-8")은 파일을 읽으려고 엽니다. 열린 파일을 for line in stream으로 돌면 한 줄씩 읽습니다. with로 연 파일은 블록이 끝날 때 오류가 나도 자동으로 닫힙니다. TemporaryDirectory는 블록 안에서만 쓰는 임시 폴더를 만들고, 블록이 끝나면 그 안의 파일까지 지웁니다. ExitStack은 개수가 실행 중에 정해지는 여러 파일을 한 블록에서 열고 함께 닫을 때 씁니다.

import heapq
from contextlib import ExitStack
from pathlib import Path
from tempfile import TemporaryDirectory


def write_run(folder, index, values):
    path = Path(folder) / f"run-{index}.txt"
    path.write_text("".join(f"{value}\n" for value in sorted(values)), encoding="utf-8")
    return path


def external_sort_file(source, destination, chunk_size):
    source, destination = Path(source), Path(destination)
    if type(chunk_size) is not int or chunk_size < 1:
        raise ValueError("청크 크기는 1 이상인 정수여야 합니다")
    if source.resolve() == destination.resolve():
        raise ValueError("원본을 덮어쓰지 않도록 출력 경로를 다르게 정해야 합니다")
    with TemporaryDirectory() as folder:
        run_paths, chunk = [], []
        with source.open(encoding="utf-8") as stream:
            for line in stream:
                chunk.append(int(line))
                if len(chunk) == chunk_size:
                    run_paths.append(write_run(folder, len(run_paths), chunk))
                    chunk = []
        if chunk:
            run_paths.append(write_run(folder, len(run_paths), chunk))
        print("런", [[int(x) for x in path.read_text(encoding="utf-8").split()] for path in run_paths])
        with ExitStack() as stack:
            streams = [stack.enter_context(path.open(encoding="utf-8")) for path in run_paths]
            runs = [(int(line) for line in stream) for stream in streams]
            with destination.open("w", encoding="utf-8") as output:
                for value in heapq.merge(*runs):
                    output.write(f"{value}\n")


with TemporaryDirectory() as folder:
    source = Path(folder) / "input.txt"
    destination = Path(folder) / "output.txt"
    source.write_text("8\n1\n6\n3\n9\n2\n7\n", encoding="utf-8")
    external_sort_file(source, destination, chunk_size=3)
    print([int(line) for line in destination.read_text(encoding="utf-8").split()])
    try:
        external_sort_file(source, source, chunk_size=3)
    except ValueError as error:
        print("오류:", error)

런 [[1, 6, 8], [2, 3, 9], [7]]
[1, 2, 3, 6, 7, 8, 9]
오류: 원본을 덮어쓰지 않도록 출력 경로를 다르게 정해야 합니다

첫 단계의 반복은 입력을 한 줄씩 읽어 청크에 모으고, 청크가 가득 차면 정렬해 런 파일로 씁니다. 확인하기 쉽도록 런 파일들의 내용을 한 번 출력했는데, 실제로 큰 자료를 다룰 때는 이 줄을 빼야 합니다. 런을 모두 메모리로 읽어 들이는 일이기 때문입니다.

두 번째 단계의 (int(line) for line in stream)은 생성기 표현식(generator expression)입니다. 대괄호로 쓴 리스트 컴프리헨션과 달리, 괄호로 쓴 생성기는 값을 미리 모두 만들지 않고 다음 값이 필요할 때마다 파일에서 한 줄을 읽어 정수 하나를 내놓습니다. 그래서 runs는 런마다 생성기 하나씩, 모두 k개의 객체만 담고 있습니다. heapq.merge(*runs)의 *runs는 리스트의 원소들을 각각의 인자로 넘기는 문법이고, heapq.merge는 출력할 값을 요청받을 때마다 가장 작은 맨 앞 값을 내놓고 그 런에서 다음 값을 하나 읽습니다. 출력 파일을 입력 파일과 같은 경로로 주면 원본이 지워지므로 함수가 거절합니다.

왜 맞고 메모리는 얼마나 드는가

첫 단계에서 만든 런은 각각 sorted로 정렬했으므로 오름차순입니다. 두 번째 단계에서는 늘 남은 값 가운데 가장 작은 값을 출력하고, 값을 꺼낸 런에서만 한 칸 나아가므로 값을 잃거나 두 번 쓰지 않습니다. 출력할 때마다 남은 값이 하나씩 줄어 모든 런이 소진되면 끝납니다. 이것은 병합 정렬의 병합과 같은 논리를 런 k개로 넓힌 것입니다.

메모리는 첫 단계에서 청크 하나, 곧 값 M개를 씁니다. 두 번째 단계에서는 런마다 맨 앞 값 하나와 열린 파일 하나가 필요하므로 런의 수 k에 비례하는 메모리를 씁니다. 파일을 한 줄씩 읽는다고 해서 디스크를 한 글자씩 읽는 것은 아니고, 파일 객체는 효율을 위해 어느 정도 크기의 버퍼(buffer)에 미리 읽어 둡니다. 그래서 실제로는 런 k개의 버퍼도 메모리를 차지합니다.

런이 너무 많을 때

값이 N개이고 청크 크기가 M이면 런은 약 N / M개입니다. 예를 들어 값이 10억 개이고 메모리에 100만 개가 들어간다면 런이 1000개 생깁니다. 운영체제가 한 프로그램에 허락하는 열린 파일의 수에는 한도가 있고, 런마다 버퍼도 필요하므로 런 1000개를 한꺼번에 여는 것은 어려울 수 있습니다. 이때는 한 번에 합치는 런의 수, 곧 병합 폭(merge width)을 정해 두고 여러 번에 나누어 병합합니다. 병합 폭이 10이면 런 10개씩 합쳐 런 100개를 만들고, 다시 10개씩 합쳐 10개, 다시 합쳐 1개를 만듭니다.

import heapq
import random


def merge_passes(runs, width):
    passes = 0
    while len(runs) > 1:
        runs = [list(heapq.merge(*runs[i:i + width])) for i in range(0, len(runs), width)]
        passes += 1
    return runs[0] if runs else [], passes


rng = random.Random(2026)
values = [rng.randrange(10**6) for _ in range(1000)]
runs = [sorted(values[i:i + 10]) for i in range(0, len(values), 10)]
for width in [100, 10, 4, 2]:
    result, passes = merge_passes(runs, width)
    print(width, result == sorted(values), passes)

100 True 1
10 True 2
4 True 4
2 True 7

이 코드는 파일 대신 리스트로 런 100개를 흉내 내어, 병합 폭에 따라 병합을 몇 번 거쳐야 하는지 셉니다. 병합 폭이 p이면 한 번 거칠 때마다 런의 수가 1/p로 줄므로, 런 k개를 하나로 만드는 데 약 log_p k번을 거칩니다. 병합을 한 번 거칠 때마다 모든 값을 파일에서 읽고 다시 써야 하는데, 메모리 계층에서 보듯 디스크를 읽고 쓰는 일은 메모리 안의 비교보다 훨씬 느립니다. 그래서 외부 정렬에서는 비교 횟수보다 자료 전체를 몇 번 읽고 쓰는지가 성능을 좌우하고, 메모리가 허락하는 한 병합 폭을 크게 잡아 거치는 횟수를 줄입니다. 파일 셋으로 런을 번갈아 옮기며 병합하는 방법은 선택 심화: 세 파일 polyphase 병합의 분배와 역할 교대에서 다룹니다.

비용

값이 N개, 청크 크기가 M, 런이 k개일 때, 첫 단계는 청크마다 정렬하므로 비교가 O(N log M)이고, 한 번의 다중 병합은 값마다 힙 연산을 하므로 O(N log k)입니다. 메모리 안의 계산을 모두 합하면 O(N log M + N log k), 곧 O(N log N)으로 내부 정렬과 같은 규모입니다. 하지만 실제 시간은 디스크를 읽고 쓰는 양에 좌우되며, 그 양은 자료 전체 N에 병합을 거치는 횟수를 곱한 정도입니다. 메모리는 청크의 M개와 런마다의 맨 앞 값과 버퍼를 합한 만큼입니다.

더 깊이: 런을 더 길게 만드는 교체 선택

런이 길수록 런의 수가 줄고, 병합 폭이 같다면 병합을 거치는 횟수도 줄어듭니다. 메모리 크기 M을 그대로 두고도 런을 더 길게 만드는 방법으로 교체 선택(replacement selection)이 있습니다. 메모리에 값 M개를 최소 힙으로 두고, 가장 작은 값을 런에 쓴 뒤 그 자리에 입력의 다음 값을 넣습니다. 새 값이 방금 쓴 값 이상이면 지금 런에 이어서 쓸 수 있으므로 힙에 넣고, 더 작으면 다음 런으로 미룹니다. 힙에 지금 런에 쓸 값이 하나도 남지 않으면 런을 끝내고, 미뤄 둔 값들로 다음 런을 시작합니다.

이렇게 하면 무작위 입력에서 런의 평균 길이가 메모리 크기의 약 두 배인 2M이 된다는 것이 알려져 있습니다. 무작위 값 10만 개를 메모리 크기 1000으로 교체 선택하면 런이 51개 생기고 평균 길이가 약 1961입니다. 청크를 정렬하는 방법이었다면 런이 100개였을 것입니다. 입력이 이미 거의 정렬되어 있다면 런이 훨씬 더 길어져, 전체가 런 하나로 끝나기도 합니다.

정리

• 외부 정렬은 메모리보다 큰 자료를, 메모리에 들어가는 청크씩 정렬해 런 파일로 만든 뒤, 런들을 하나로 합치는 두 단계로 정렬합니다.
• 다중 병합은 런마다 맨 앞 값 하나만 메모리에 두고, 최소 힙으로 가장 작은 값을 골라 출력한 뒤 그 런에서만 다음 값을 읽습니다.
• Python의 heapq.merge와 생성기 표현식을 쓰면, 런 전체를 메모리에 올리지 않고 필요할 때마다 한 줄씩 읽으며 합칠 수 있습니다.
• 런이 너무 많으면 병합 폭을 정해 여러 번에 나누어 병합하며, 병합 폭이 p일 때 약 log_p k번을 거칩니다.
• 외부 정렬의 성능은 비교 횟수보다 자료 전체를 몇 번 읽고 쓰는지가 좌우하므로, 런을 길게 만들고 병합 폭을 크게 잡아 거치는 횟수를 줄입니다.

연습 문제

1. 청크 크기를 바꾸면

청크 크기를 2로 바꾸면 본문의 입력 8, 1, 6, 3, 9, 2, 7에서 어떤 런들이 만들어지나요?

2. 런이 백만 개라면

런이 백만 개 생겼을 때 본문의 external_sort_file을 그대로 쓰기 어려운 이유를 두 가지 들고, 어떻게 해결하면 될지 설명하세요.

3. 병합을 거치는 횟수

런이 1000개이고 병합 폭이 10이면 병합을 몇 번 거쳐야 하나요? 병합 폭을 32로 늘리면 어떻게 될까요?

다음 글 선택 심화: 단절점·low·이중 연결 성분에서는 다시 그래프로 돌아가, 깊이 우선 탐색으로 그래프의 약한 연결 지점을 찾습니다.