(CS) 메모리보다 큰 데이터를 정렬하는 방법

(CS) 메모리보다 큰 데이터를 정렬하는 방법 (이론편)

배경

개발을 할때 항상 중요하게 생각하는 부분이 있습니다. 바로 시스템의 규모입니다.

우리가 1+1이라는 기능을 개발해도 이 기능을 100명이 쓰는거랑 100만명이 사용하는 것은 아주 큰 차이가 존재합니다.

만약 규모가 작은 환경이라면, 대규모 아키텍처를 깊이 고려하지 않아도 서비스에 지장이 없습니다.

하지만.. 규모가 커진다면 고민은 깊어질 것입니다.

같은 정렬이라도. 대규모 환경에서 마주할법한 고민이라도 생각하여 스터디하게 되었습니다.

기본 원리

우리가 익히 알고있는 대부분의 정렬 알고리즘은 데이터를 리스트 혹은 배열 형태로 메모리에 올리고 알고리즘에 맞게 풀스캔을 진행 하면서 데이터를 서로 스위칭 하는 과정을 통해 정렬을 합니다.

당연히 메모리는 환경에 따라 한계가 존재하기 때문에 만약 대상 데이터가 메모리보다 큰 용량을 가지고 있다면, 모든 데이터를 하나의 자료구조에 넣기는 불가능할 것입니다.

메모리에 모든 데이터를 올릴 수 없다면 자연스럽게 생각 나는 방법은 부분적으로 정렬해서 합쳐야겠다는 생각이 떠오릅니다.

주기억장치가 읽을 수 있는 블럭 만큼을 보조 기억 장치에서 읽어서 해당 크기만큼 정렬을 진행한 후 다시 보조 기억장치에 저장합니다.

블럭별로 정렬이 완료 되었다면 각 블럭들을 부분적으로 읽어서 합병하여 전체를 정렬하면 메모리가 부족해도 정렬이 가능할 것입니다. (자세한 병합 방법은 아래에)

이 과정은 merge sort(합병 정렬)와 과정이 동일합니다.

이처럼 전체 데이터를 메모리에 한 번에 적재하지 않고, 디스크에 저장된 데이터를 여러 블록으로 나누어 정렬한 뒤 병합하는 방식을 External Merge Sort 이라고 합니다.

Merge Sort(병합 정렬)

병합 정렬 알고리즘은 분할정복을 활용하여 정렬을 진행합니다.

구체적으로 다음과 같이 진행됩니다.

• 하나의 리스트를 두 개의 균등한 크기로 분할하고 분할된 부분 리스트를 정렬한 다음, 두 개의 정렬된 부분 리스트를 합하여 전체가 정렬된 리스트가 되게 합니다.
  • 분할(Divide): 입력 배열을 같은 크기의 2개의 부분 배열로 분할
  • 정복(Conquer): 부분 배열을 정렬하고 부분 배열의 크기가 충분히 작지 않으면 순환 호출 을 이용하여 다시 분할 정복 방법을 적용
  • 결합(Combine): 정렬된 부분 배열들을 하나의 배열에 합병

아래 사진은 합병 정렬 과정을 나타낸 것입니다.

External Merge Sort

• MySQL에서도 사용하고 있는 정렬 방식입니다.

• 메모리 크기 만큼 읽어서 정렬된 run을 만들고 그 run들을 병합합니다.

상세 동작

M = 메모리에 한번에 올릴 수 있는 크기
N = 전체 데이터 크기

1. 보조 메모리에서 M 크기 많큼 읽어서 정렬하고 저장합니다.
   이때 저장한 단위가 Run 이라는 단위가 되고 N/M번 반복하여 Run을 만듭니다.

2. 메모리 버퍼 수(B개) 만큼의 블록을 동시에 열어서 병합합니다.

   • 각 run에서 현재 포인터 값만 메모리에서 유지합니다.
   • 최소 Heap을 사용해서 각 run에서 읽은 최소 값들을 꺼냅니다.
   • 최소 값을 꺼내고 결과에 이어 붙입니다.
   • 값을 꺼낸 run의 다음 값을 Heap에 넣습니다.

병합 예시

Initial Runs
--------------------------------------------------
Run A: [5  8  9]
Run B: [1  6  7]
Run C: [2  4 10]

초기 Heap 상태 (각 run의 첫 값)
Heap: 5(A), 1(B), 2(C)
Output: []

--------------------------------------------------
Step 1
min = 1 (B)
Output: [1]
Heap: 5(A), 6(B), 2(C)

--------------------------------------------------
Step 2
min = 2 (C)
Output: [1, 2]
Heap: 5(A), 6(B), 4(C)

--------------------------------------------------
Step 3
min = 4 (C)
Output: [1, 2, 4]
Heap: 5(A), 6(B), 10(C)

--------------------------------------------------
Step 4
min = 5 (A)
Output: [1, 2, 4, 5]
Heap: 8(A), 6(B), 10(C)

--------------------------------------------------
Step 5
min = 6 (B)
Output: [1, 2, 4, 5, 6]
Heap: 8(A), 7(B), 10(C)

--------------------------------------------------
Step 6
min = 7 (B)
Output: [1, 2, 4, 5, 6, 7]
Heap: 8(A), 10(C)

--------------------------------------------------
Step 7
min = 8 (A)
Output: [1, 2, 4, 5, 6, 7, 8]
Heap: 9(A), 10(C)

--------------------------------------------------
Step 8
min = 9 (A)
Output: [1, 2, 4, 5, 6, 7, 8, 9]
Heap: 10(C)

--------------------------------------------------
Step 9
min = 10 (C)
Output: [1, 2, 4, 5, 6, 7, 8, 9, 10]

Final Result
--------------------------------------------------
[1, 2, 4, 5, 6, 7, 8, 9, 10]

Binary Sort/Merge (2-Way Merge)

• 2개씩 진행하기 때문에 효율적으로 사용하지 못할 수 있음

K-Way Merge

최대한 메모리에서 수용 가능한 파일 수를 활용하여서 병합합니다.

간단 실습

• 전체 데이터: 100개의 랜덤 정수

• run 크기: 3

• heap 크기(= 병합 시 메모리에 올릴 수 있는 값 수): 5

code

import java.util.*;

public class ExternalMergeSort {

    static final int TOTAL_NUMBERS = 100;
    static final int RUN_SIZE = 3;      // run 크기
    static final int MERGE_WAYS = 5;    // heap 크기 (동시에 병합 가능한 run 개수)

    public static void main(String[] args) throws Exception {

        String inputFile = "input.txt";

        generateRandomFile(inputFile);
        List<File> runs = createInitialRuns(inputFile);

        int pass = 0;
        while (runs.size() > 1) {
            runs = mergePass(runs, pass++);
        }

        System.out.println("정렬 완료: " + runs.get(0).getName());
        printFile(runs.get(0));
    }

    // 랜덤 파일 생성
    static void generateRandomFile(String filename) throws IOException {
        Random rand = new Random();
        try (BufferedWriter bw = new BufferedWriter(new FileWriter(filename))) {
            for (int i = 0; i < TOTAL_NUMBERS; i++) {
                bw.write(rand.nextInt(1000) + "\n");
            }
        }
    }

    // 초기 run 생성
    static List<File> createInitialRuns(String inputFile) throws IOException {
        List<File> runs = new ArrayList<>();

        try (BufferedReader br = new BufferedReader(new FileReader(inputFile))) {
            List<Integer> buffer = new ArrayList<>();
            String line;
            int runCount = 0;

            while ((line = br.readLine()) != null) {
                buffer.add(Integer.parseInt(line));

                if (buffer.size() == RUN_SIZE) {
                    runs.add(writeRun(buffer, runCount++));
                    buffer.clear();
                }
            }

            if (!buffer.isEmpty()) {
                runs.add(writeRun(buffer, runCount));
            }
        }

        return runs;
    }

    static File writeRun(List<Integer> data, int runIndex) throws IOException {
        Collections.sort(data);
        File runFile = new File("run_" + runIndex + ".txt");

        try (BufferedWriter bw = new BufferedWriter(new FileWriter(runFile))) {
            for (int num : data) {
                bw.write(num + "\n");
            }
        }

        return runFile;
    }

    // Merge Pass
    static List<File> mergePass(List<File> runs, int pass) throws IOException {
        List<File> newRuns = new ArrayList<>();
        int index = 0;
        int newRunIndex = 0;

        while (index < runs.size()) {
            int end = Math.min(index + MERGE_WAYS, runs.size());
            List<File> subset = runs.subList(index, end);
            File merged = mergeRuns(subset, pass, newRunIndex++);
            newRuns.add(merged);
            index = end;
        }

        return newRuns;
    }

    // K-way merge
    static File mergeRuns(List<File> runFiles, int pass, int runIndex) throws IOException {

        PriorityQueue<Node> pq = new PriorityQueue<>(Comparator.comparingInt(n -> n.value));
        List<BufferedReader> readers = new ArrayList<>();

        for (int i = 0; i < runFiles.size(); i++) {
            BufferedReader br = new BufferedReader(new FileReader(runFiles.get(i)));
            readers.add(br);
            String line = br.readLine();
            if (line != null) {
                pq.add(new Node(Integer.parseInt(line), i));
            }
        }

        File output = new File("pass_" + pass + "_run_" + runIndex + ".txt");

        try (BufferedWriter bw = new BufferedWriter(new FileWriter(output))) {

            while (!pq.isEmpty()) {
                Node node = pq.poll();
                bw.write(node.value + "\n");

                String next = readers.get(node.runIndex).readLine();
                if (next != null) {
                    pq.add(new Node(Integer.parseInt(next), node.runIndex));
                }
            }
        }

        for (BufferedReader br : readers) {
            br.close();
        }

        return output;
    }

    static void printFile(File file) throws IOException {
        try (BufferedReader br = new BufferedReader(new FileReader(file))) {
            String line;
            while ((line = br.readLine()) != null) {
                System.out.print(line + " ");
            }
        }
    }

    static class Node {
        int value;
        int runIndex;

        Node(int value, int runIndex) {
            this.value = value;
            this.runIndex = runIndex;
        }
    }
}

Reference

• https://gmlwjd9405.github.io/2018/05/08/algorithm-merge-sort.html
• https://velog.io/@ilil1/Chapter-1.-%EC%99%B8%EB%B6%80-%EC%A0%95%EB%A0%ACExternal-sort-Sorting-4
• https://ybdeveloper.tistory.com/118