4.6 배치 변환과 자동화 - 진행률, 병렬 처리, 결과 검증

#셸 스크립트에서 서비스 코드로

명령을 손으로 실행하는 단계를 넘어 MAM의 작업 큐에서 FFmpeg를 호출하려면, 다음 문제를 해결해야 합니다.

1. 진행률을 사용자에게 보여 줘야 한다.
2. 변환이 중간에 실패했을 때 불완전한 파일이 결과물로 남으면 안 된다.
3. 동시에 몇 개를 실행할지 통제해야 한다.
4. 변환이 끝난 파일이 정말 정상인지 검증해야 한다.
5. 같은 작업이 중복 실행되어도 결과가 망가지지 않아야 한다.

#진행률: -progress 옵션

사람이 보는 frame= ... speed= 줄을 파싱하는 방법은 포맷이 바뀔 수 있어 불안정합니다. FFmpeg는 기계가 읽기 위한 -progress 출력을 제공합니다.

ffmpeg -v error -nostats -progress pipe:1 -i src10.mxf -c:v libx264 -preset ultrafast -an pg.mp4

출력은 키=값 형태로 주기적으로 나옵니다.

frame=300
out_time_us=10010000
speed=15.6x
progress=end

• out_time_us: 지금까지 출력한 시간(마이크로초). 원본 길이로 나누면 진행률입니다.
• progress: 중간에는 continue, 마지막에는 end.
• speed: 실시간 대비 속도. 남은 시간 예측에 씁니다.

원본의 길이는 변환 전에 ffprobe로 구해 둡니다.

#안전한 변환 함수

아래는 위 문제들을 처리한 파이썬 예제입니다. 이름이 같은 a.mxf와 a.mov, b.mxf, 그리고 일부러 만든 깨진 파일 broken.mxf를 3개씩 동시에 변환해 확인했습니다.

import json
import os
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

PROXY_ARGS = ['-map', '0:v:0', '-map', '0:a:0?', '-vf', 'yadif,scale=-2:360',
              '-c:v', 'libx264', '-preset', 'fast', '-crf', '28', '-g', '30',
              '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-b:a', '96k',
              '-movflags', '+faststart']


def probe_duration(path):
    out = subprocess.run(['ffprobe', '-v', 'error', '-show_entries', 'format=duration',
                          '-of', 'json', str(path)], capture_output=True, text=True, check=True).stdout
    value = json.loads(out)['format'].get('duration')
    return float(value) if value not in (None, 'N/A') else None


def dst_for(dst_dir, src):
    # a.mxf 와 a.mov 가 같은 결과 이름을 쓰지 않도록 원본 확장자를 이름에 포함한다
    return dst_dir / f'{src.stem}_{src.suffix.lstrip(".").lower()}_proxy.mp4'


def convert(src, dst):
    lock = dst.with_name(dst.name + '.lock')
    try:
        os.close(os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY))   # 같은 대상을 두 곳에서 동시에 만들지 않는다
    except FileExistsError:
        return src.name, 'SKIP (다른 작업이 처리 중)'
    try:
        try:
            total = probe_duration(src)
        except subprocess.CalledProcessError:
            return src.name, 'FAIL (읽을 수 없는 파일)'
        if total is None:
            return src.name, 'FAIL (길이를 알 수 없음)'
        tmp = dst.with_name(dst.stem + '.part' + dst.suffix)
        cmd = ['ffmpeg', '-v', 'error', '-nostdin', '-y', '-progress', 'pipe:1', '-nostats',
               '-i', str(src), *PROXY_ARGS, str(tmp)]
        proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True)
        for line in proc.stdout:
            if line.startswith('out_time_us='):
                value = line.split('=')[1].strip()
                if value.isdigit():
                    print(f'{src.name}: {min(100, int(value) / 1e6 / total * 100):5.1f}%', flush=True)
        if proc.wait() != 0:
            tmp.unlink(missing_ok=True)
            return src.name, 'FAIL (ffmpeg exit code)'
        try:
            out_len = probe_duration(tmp)
        except subprocess.CalledProcessError:
            out_len = None                               # 결과 파일을 읽을 수 없으면 실패
        if out_len is None or abs(out_len - total) > 0.5:      # 길이 검증
            tmp.unlink(missing_ok=True)
            return src.name, 'FAIL (duration mismatch)'
        os.replace(tmp, dst)                             # 검증 통과 후에만 최종 이름으로
        return src.name, 'OK'
    finally:
        lock.unlink(missing_ok=True)


if __name__ == '__main__':
    src_dir, dst_dir, workers = Path(sys.argv[1]), Path(sys.argv[2]), int(sys.argv[3])
    dst_dir.mkdir(exist_ok=True)
    jobs = [(p, dst_for(dst_dir, p)) for p in sorted(src_dir.iterdir())
            if p.suffix.lower() in ('.mxf', '.mov', '.mp4', '.ts')
            and not dst_for(dst_dir, p).exists()]
    with ThreadPoolExecutor(max_workers=workers) as pool:
        for name, status in pool.map(lambda j: convert(*j), jobs):
            print(f'RESULT {name}: {status}')

RESULT a.mov: OK
RESULT a.mxf: OK
RESULT b.mxf: OK
RESULT broken.mxf: FAIL (읽을 수 없는 파일)
(결과 파일: a_mov_proxy.mp4, a_mxf_proxy.mp4, b_mxf_proxy.mp4)
(다시 실행하면 완료된 3개는 건너뛰고 broken.mxf만 다시 FAIL로 보고)
(같은 폴더에 두 프로세스를 동시에 실행하면 나중 프로세스는 처리 중인 파일을 SKIP)
(결함 주입 테스트: 결과 파일을 ffprobe가 읽지 못하는 경우 → FAIL (duration mismatch), 입력의 길이가 N/A인 경우 → FAIL (길이를 알 수 없음). 두 경우 모두 최종 파일과 .part·.lock 파일이 남지 않음)

#설계 포인트

#병렬 처리

변환은 CPU(또는 GPU)를 많이 쓰므로 동시 작업 수 제한이 중요합니다.

• x264는 기본적으로 여러 코어를 모두 사용하므로, 작업을 코어 수만큼 동시에 돌리면 오히려 서로 느려집니다. 보통 동시 작업 수 = 코어 수 ÷ 인코더 스레드 수 정도에서 시작해 측정으로 조정합니다.
• 디스크 I/O가 병목이 되는 경우가 많습니다. 원본이 네트워크 스토리지에 있다면 동시에 읽는 작업 수를 대역폭에 맞춰 제한합니다.
• 우선순위가 다른 작업(긴급 송출용 vs 일반 프록시)은 큐를 분리하세요.

하드웨어 인코더(h264_nvenc 등)는 속도가 빠르지만 같은 CRF에서 소프트웨어 인코더(libx264)보다 효율이 낮고 옵션 체계가 달라서, 프록시용으로는 속도, 최종 배포용으로는 libx264처럼 용도를 나누어 쓰는 경우가 많습니다.

#로그와 재현성

운영 환경에서는 실패했을 때 정확히 어떤 명령을 실행했는지 알 수 있어야 합니다.

• 실행한 전체 명령줄, FFmpeg 버전, 종료 코드, 표준 오류 출력을 작업 이력에 저장합니다.
• 변환 프로파일의 이름과 버전을 함께 기록해 두면 "같은 소재를 같은 설정으로 다시 변환"할 수 있습니다.
• -v error로 로그를 줄이되, 실패한 작업은 -v warning이나 -v info로 다시 실행해 원인을 확인할 수 있도록 합니다.

#정리

• 진행률은 -progress pipe:1의 out_time_us로 계산합니다.
• 결과는 임시 파일에 쓰고, 종료 코드와 길이를 검증한 뒤에만 최종 이름으로 바꿉니다.
• 동시 작업 수, 큐 분리, 실행 명령 기록으로 운영 안정성을 확보합니다.

#참고 자료

• FFmpeg 문서: ffmpeg 명령 옵션

링크는 2026-10-28에 열람이 되는 것을 확인했습니다. 규격과 도구 문서는 개정될 수 있으므로 적용하기 전에 최신 판을 확인하세요.