병렬 병합 정렬과 OpenMP·MPI·MapReduce 선택 기준

정렬 작업을 병렬화할 때 먼저 정해야 할 것

대용량 데이터의 정렬과 배치·스트리밍 처리는 분할 방식, 병합 경로, 실행 환경에 따라 병목의 위치가 달라진다. 병렬 병합 정렬은 분할 정복 방식의 병합 정렬을 태스크 또는 데이터 병렬성으로 확장해 코어와 노드에서 동시에 수행하는 기법이다.

OpenMP는 단일 노드의 공유 메모리를 대상으로 한 병렬화 표준이고, MPI는 여러 노드의 분산 메모리 환경에서 메시지를 주고받는 표준이다. MapReduce는 키-값 데이터를 맵 단계에서 분산 처리하고, 셔플 정렬과 리듀스 집계를 거쳐 대규모 처리를 수행하는 프레임워크다.

분할과 병합이 전체 구조를 결정한다

병렬 병합 정렬에서는 입력을 어느 단위로 나눌지 정하고, 각 구간을 로컬에서 정렬한 뒤 다중 병합으로 결과를 합친다. 이때 작업 큐 기반의 동적 스케줄링, 태스크 종속성, 병합 단계의 배치가 함께 고려 대상이 된다.

하향식 분할은 캐시 친화적인 서브배열을 만들 수 있으며, 병합 버퍼를 재사용하면 메모리 비용을 조절할 수 있다. OpenMP에서는 재귀 태스크 생성, 병합 시점의 동기화, cut-off 깊이가 핵심이다. MPI에서는 Scatter/Gatherv를 이용한 블록 분산과 수집, 트리형 pairwise 병합 또는 루트 k-way 병합을 설계한다.

MapReduce의 전역 정렬은 샘플링으로 파티션 경계를 정하고 TotalOrderPartitioner로 데이터를 배치하는 방식에 기반한다. 맵 출력 키 정렬, 파티션 라우팅, 리듀서 내부 병합 정렬이 이어진다.

⤢✕

입력은 배열 또는 레코드 N이며, 처리 경로는 분할·로컬 정렬·병합 또는 Map·Shuffle·Reduce로 나뉜다. 스레드가 과도하게 생성되면 cut-off를 상향 조정한다. MPI 노드 실패에는 작업 중단 또는 체크포인트 복구가 필요하고, MapReduce 태스크 실패는 프레임워크가 자동 재시도한다.

공유 메모리에서 태스크로 병합 정렬 실행하기

OpenMP 구현에는 g++ 또는 clang++와 -fopenmp를 지원하는 환경이 필요하다.

컴파일: g++ -O2 -fopenmp pms_omp.cpp -o pms_omp

// pms_omp.cpp
#include <bits/stdc++.h>
#include <omp.h>
using namespace std;

static const size_t CUTOFF = 1<<14; // 16K 원소 단위 cut-off

void merge_vec(vector<int>& a, size_t l, size_t m, size_t r, vector<int>& buf) {
    size_t i=l, j=m, k=l;
    while (i<m && j<r) buf[k++] = (a[i] <= a[j]) ? a[i++] : a[j++];
    while (i<m) buf[k++] = a[i++];
    while (j<r) buf[k++] = a[j++];
    for (size_t t=l; t<r; ++t) a[t] = buf[t];
}

void pmerge_sort(vector<int>& a, size_t l, size_t r, vector<int>& buf, int depth=0) {
    size_t n = r - l;
    if (n <= 1) return;
    if (n < CUTOFF) {
        sort(a.begin()+l, a.begin()+r);
        return;
    }
    size_t m = l + n/2;
    if (depth < 20) {
        #pragma omp task shared(a, buf) firstprivate(l, m, depth)
        pmerge_sort(a, l, m, buf, depth+1);
        #pragma omp task shared(a, buf) firstprivate(m, r, depth)
        pmerge_sort(a, m, r, buf, depth+1);
        #pragma omp taskwait
    } else {
        pmerge_sort(a, l, m, buf, depth+1);
        pmerge_sort(a, m, r, buf, depth+1);
    }
    merge_vec(a, l, m, r, buf);
}

int main(int argc, char** argv){
    size_t N = (argc>1)? stoull(argv[1]) : 1<<20; // 기본 1M
    vector<int> a(N), buf(N);
    mt19937 rng(42); uniform_int_distribution<int> dist(INT_MIN, INT_MAX);
    for (auto &x: a) x = dist(rng);

    double t0 = omp_get_wtime();
    #pragma omp parallel
    {
        #pragma omp single nowait
        pmerge_sort(a, 0, a.size(), buf);
    }
    double t1 = omp_get_wtime();

    // 검증
    if (!is_sorted(a.begin(), a.end())) { cerr << "정렬 실패\n"; return 1; }
    cout << "N=" << N << " 시간(초)=" << fixed << setprecision(3) << (t1-t0)
         << " 스레드=" << omp_get_max_threads() << "\n";
    return 0;
}

조정 대상은 cut-off 크기, task depth 제한, 병합 버퍼 재사용, NUMA 바인딩(OMP_PROC_BIND)이다.

분산 메모리 환경에서 데이터와 병합을 배치하기

MPI 구현은 OpenMPI 또는 MPICH를 설치한 뒤 단일 실행 파일을 mpirun -n P로 실행한다.

컴파일: mpicc -O2 pms_mpi.c -o pms_mpi

// pms_mpi.c
#include <mpi.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

int cmp_int(const void* a, const void* b){ int x=*(int*)a, y=*(int*)b; return (x>y)-(x<y); }

// 루트에서 k개의 정렬된 블록을 k-way 병합
int* kway_merge(int* data, int* counts, int k, int* offsets, int total){
    int* idx = (int*)calloc(k, sizeof(int));
    int* out = (int*)malloc(sizeof(int)*total);
    for(int i=0;i<total;i++){
        int sel=-1, sel_val=0;
        for(int j=0;j<k;j++){
            int pos = idx[j];
            if(pos < counts[j]){
                int val = data[offsets[j]+pos];
                if(sel<0 || val < sel_val){ sel=j; sel_val=val; }
            }
        }
        out[i] = sel_val;
        idx[sel]++;
    }
    free(idx);
    return out;
}

int main(int argc, char** argv){
    MPI_Init(&argc,&argv);
    int rank, size; MPI_Comm_rank(MPI_COMM_WORLD,&rank); MPI_Comm_size(MPI_COMM_WORLD,&size);

    int N = (argc>1)? atoi(argv[1]) : (1<<20);
    int *root_data = NULL, *counts=NULL, *displs=NULL;

    if(rank==0){
        root_data = (int*)malloc(sizeof(int)*N);
        // 데이터 생성
        srand(42);
        for(int i=0;i<N;i++) root_data[i] = rand();
        counts = (int*)malloc(sizeof(int)*size);
        displs = (int*)malloc(sizeof(int)*size);
        int base = N/size, rem = N%size;
        int off=0;
        for(int p=0;p<size;p++){
            counts[p] = base + (p<rem);
            displs[p] = off; off += counts[p];
        }
    }

    // 각 랭크 수신 크기 브로드캐스트
    int mycount;
    if(rank==0) mycount = counts[0];
    MPI_Scatter(counts, 1, MPI_INT, &mycount, 1, MPI_INT, 0, MPI_COMM_WORLD);

    int* local = (int*)malloc(sizeof(int)*mycount);
    // 가변 분산
    MPI_Scatterv(root_data, counts, displs, MPI_INT, local, mycount, MPI_INT, 0, MPI_COMM_WORLD);

    // 로컬 정렬
    qsort(local, mycount, sizeof(int), cmp_int);

    // 루트로 수집
    MPI_Gatherv(local, mycount, MPI_INT, root_data, counts, displs, MPI_INT, 0, MPI_COMM_WORLD);

    if(rank==0){
        // 루트에서 k-way 병합
        double t0 = MPI_Wtime();
        int* sorted = kway_merge(root_data, counts, size, displs, N);
        double t1 = MPI_Wtime();
        // 검증
        for(int i=1;i<N;i++){
            if(sorted[i-1] > sorted[i]){ fprintf(stderr,"정렬 실패\n"); MPI_Abort(MPI_COMM_WORLD,1); }
        }
        printf("N=%d, ranks=%d, 병합시간=%.3f초\n", N, size, t1-t0);
        free(sorted);
        free(root_data); free(counts); free(displs);
    }

    free(local);
    MPI_Finalize();
    return 0;
}

균등 분할에 샘플링 기반 파티셔닝(PSRS, sample sort)을 더하면 부하 균형을 개선할 수 있다. 루트 병합의 병목은 트리형 pairwise 병합으로 완화하며, 비차단 통신(Irecv/Isend)은 통신과 다른 작업의 오버랩에 사용한다.

MapReduce로 전역 키 순서를 유지하는 방식

Hadoop 3.x Streaming과 Python 3 예제는 Hadoop 클러스터 및 HDFS 접근 권한을 전제로 한다. 전역 정렬 보장에는 TotalOrderPartitioner를 사용한다.

mapper.py

#!/usr/bin/env python3
import sys
for line in sys.stdin:
    key = line.rstrip("\n")
    print(f"{key}\t")

reducer.py는 identity reducer다.

#!/usr/bin/env python3
import sys
for line in sys.stdin:
    # 키 정렬은 셔플 단계에서 보장
    sys.stdout.write(line.split("\t",1)[0] + "\n")

샘플링 파일을 먼저 생성한다.

• 입력 경로: /data/nums (각 라인 숫자)
• 샘플러로 파티션 경계 생성
  hadoop jar $HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-examples-*.jar
  teragen 1000000 /tmp/tera_in # 예시(테라젠 사용 시)
  혹은 커스텀 샘플러: hadoop jar ... sampl e -r 4 -in /data/nums -out /tmp/samples

파티션 파일을 만든 뒤 배치한다.

hadoop org.apache.hadoop.examples.TotalOrderPartitioner \
 -writePartitionFile jobConf /tmp/partition.lst # 최신 스크립트/옵션은 배포판별 상이, 최신 정보 확인 필요

스트리밍 잡은 리듀서=R개로 실행하며 전역 정렬 결과를 만든다.

hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-\*.jar \
 -D mapreduce.job.reduces=4 \
 -D mapreduce.partitioner.class=org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner \
 -D mapreduce.totalorderpartitioner.path=/tmp/partition.lst \
 -input /data/nums -output /out/sorted \
 -mapper mapper.py -reducer reducer.py \
 -file mapper.py -file reducer.py

전역 정렬의 품질은 파티션 경계의 정확도에 달려 있다. 입력 분포가 비균질하면 추가 샘플링이 필요하며, 리듀서 수와 파티션 경계 수도 일치해야 한다.

환경별로 달라지는 적용 지점

단일 서버의 초대용량 로그 정렬에는 NVMe SSD와 32~64 코어 서버에서 OpenMP 병렬 병합 정렬을 적용해 ingest 파이프라인의 전처리를 가속할 수 있다.

HPC 환경의 대규모 배열·행렬 인덱스 정렬에는 MPI 기반 PSRS(sample sort)로 균등 파티션을 만들고 계층 병합을 적용한다. 이 구조는 노드 수 10^2~10^3 규모까지 확장한다.

데이터 레이크의 전역 키 정렬 파이프라인에서는 Hadoop/Spark MapReduce 계열 프레임워크에 TotalOrderPartitioner를 적용해 파티션별 정렬을 유지하고 다운스트림 range join을 최적화한다.

성능 지표와 운영상 차이

단일 노드 OpenMP는 메모리 대역폭 한계 전까지 32코어 기준, 데이터형 int/long 및 NUMA 최적화를 가정하면 10–25배 속도 향상을 기대한다. MPI 클러스터는 통신·병합 오버헤드를 제외하면 거의 선형으로 확장하며, 64노드에서 40–55배 가속 범위다. MapReduce는 페타바이트급 데이터에서 선형 확장성과 SLA 기반 재시도·스케줄링 안정성을 제공한다.

스피드업은 Sp = T1/Tp, 효율은 Ep = Sp/p로 본다. Amdahl 상한은 Sp <= 1/(f + (1-f)/p)이며, Gustafson 확장성은 Sg ≈ p - α(p-1)이다. k-way 병합의 복잡도는 O(N log k), 트리형 병합 깊이는 O(log k)다.

MapReduce는 자동 재시도와 스펙큐러티브 실행으로 장애 내성을 높인다. MPI는 체크포인팅을 도입하면 가용성이 향상되며, OpenMP는 낮은 오버헤드와 단순한 운영이 장점이다.

항목	OpenMP	MPI	MapReduce
성능	노드 내 최고 성능, 낮은 동기화 오버헤드	노드 간 확장에 따른 높은 처리량	디스크/네트워크 중심, 대용량에 강함
확장성	소켓/메모리 대역폭 한계	수백~수천 노드까지 수평 확장	스토리지/컴퓨트 분리, 페타바이트급
일관성/결정성	결정적 결과, 재현 용이	결정적 결과, 통신 순서 관리 필요	TotalOrderPartitioner로 전역 정렬 보장
안정성	프로세스 내 오류 영향 큼	노드 실패 시 작업 중단 위험	태스크 재시도/스펙큐러티브 실행 지원
운영 편의	빌드 간단, 코드 변경 최소	통신/병합 설계 복잡	배포/모니터링 표준화, 튜닝 파라미터 다수

균등 분할은 구현이 단순하지만 데이터 분포가 치우치면 부하 불균형이 생긴다. 샘플링 기반 분할은 균형을 개선하는 대신 추가 패스 오버헤드를 수반한다. 루트 k-way 병합은 단순한 반면 루트 병목 위험이 있고, 트리형 pairwise 병합은 이를 줄이는 대신 통신 단계가 늘어난다.

큰 병합 버퍼는 캐시 효율을 높일 수 있지만 메모리 사용량도 증가한다. NUMA 바인딩은 로컬리티를 개선하는 대신 유연성을 낮춘다. MPI와 OpenMP는 경량·저지연이 장점이지만 장애 복구가 어렵고, MapReduce는 강한 장애 내성을 제공하는 대신 지연과 자원 오버헤드가 있다.