BioPlayground

🧬
목록으로

FASTQ 클라우드 배치 — 노트북을 벗어나 AWS로 옮기기

200GB FASTQ를 로컬에서 못 돌린다면? Docker + S3 + AWS Batch 조합으로 확장 가능한 시퀀싱 파이프라인을 파이썬으로 직접 만듭니다.

심화
|
120분
|
검증 완료 (2026-07)
FASTQ 처리AWS BatchS3클라우드 파이프라인sequencing 데이터boto3Docker
진행률0/16 (0%)

FASTQ 클라우드 배치 — 노트북을 벗어나 AWS로 옮기기

이 토픽을 마치면

교과서에서 배운 AWS 배포 와 S3 스토리지 를 엮어서, 여러분의 로컬 파이썬 스크립트를 컨테이너로 감싸 AWS Batch에서 대량 실행하는 파이프라인을 직접 만들 수 있습니다. FASTQ 파일을 예시로 삼지만, 이 패턴은 어떤 종류의 대량 데이터 처리에도 그대로 적용됩니다.

이 글은 교육용 일반 예제 입니다. 실제 프로덕션 시퀀싱 파이프라인은 Nextflow, Snakemake, WDL 같은 워크플로 언어를 씁니다.

"노트북 디스크가 꽉 찼어요" — 로컬 처리의 한계

여러분이 시퀀싱 코어에서 200GB짜리 FASTQ 파일을 받았다고 합시다. 이 파일에 대해 다음 작업을 하려 합니다.

  1. quality trimming (품질 낮은 read 제거)
  2. 참조 유전체 정렬
  3. 변이 호출

로컬 접근:

python
import subprocess


subprocess.run(["fastp", "-i", "sample.fastq.gz", "-o", "trimmed.fastq.gz"])
subprocess.run(["bwa", "mem", "reference.fa", "trimmed.fastq.gz", "-o", "aligned.sam"])
subprocess.run(["samtools", "sort", "aligned.sam", "-o", "sorted.bam"])
subprocess.run(["bcftools", "call", "sorted.bam", "-o", "variants.vcf"])

이 접근에 세 가지 실전 문제가 있습니다.

문제 1: 디스크. 원본 200GB + 중간 파일 여러 개 = 1TB 이상. 여러분의 노트북 디스크는 이 정도가 안 됩니다.

문제 2: 메모리·시간. BWA 정렬이 32GB 메모리와 몇 시간을 요구합니다. 노트북이 이 사이 다른 작업을 못 합니다.

문제 3: 확장. 이번엔 샘플 1개인데 다음 프로젝트는 100개입니다. 로컬 순차 처리로는 몇 주 걸립니다.

진짜 접근은 이 파이프라인을 컨테이너로 감싸 AWS Batch에서 실행 하는 것입니다. 원본 파일은 S3에 두고, Batch 잡이 큰 EC2 인스턴스에서 컨테이너를 띄워 파일을 처리한 뒤 결과를 다시 S3에 저장합니다. 100개 샘플이 100개의 병렬 잡으로 돌아갑니다.

블랙박스에서 부품으로 — 클라우드 파이프라인 열어보기

핵심 부품은 네 개입니다.

부품 1: S3 — 무한 저장소

S3는 사실상 무한한 용량의 객체 스토리지입니다. 파일 하나는 최대 5TB, 버킷당 파일 수 제한 없음. 라이프사이클 정책으로 오래된 파일을 자동으로 저비용 티어(Glacier)로 옮길 수 있습니다.

파이썬 SDK(boto3)로 접근합니다.

python
import boto3


s3 = boto3.client("s3")


s3.upload_file("sample.fastq.gz", "my-bucket", "raw/sample.fastq.gz")


s3.download_file("my-bucket", "results/variants.vcf", "variants.vcf")


for obj in s3.list_objects_v2(Bucket="my-bucket", Prefix="raw/")["Contents"]:
    print(obj["Key"], obj["Size"])

주의: S3 요금은 저장 + 전송 + 요청. 같은 리전 EC2에서 접근하면 전송 비용이 없습니다. 잡을 파일이 있는 리전에서 돌리는 것이 비용의 첫 원칙입니다.

부품 2: Docker 컨테이너

여러분의 파이프라인 스크립트가 필요로 하는 모든 것 — 파이썬 + fastp + bwa + samtools + bcftools + 여러분의 코드 — 을 하나의 이미지로 묶습니다.

Dockerfile 예시:

dockerfile
FROM ubuntu:22.04

RUN apt-get update && apt-get install -y \
    python3.12 python3-pip \
    fastp bwa samtools bcftools \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt /app/
RUN pip3 install -r /app/requirements.txt

COPY pipeline.py /app/
WORKDIR /app

ENTRYPOINT ["python3", "pipeline.py"]

빌드하고 ECR(Elastic Container Registry)에 push:

bash
docker build -t my-fastq-pipeline .
aws ecr get-login-password | docker login --username AWS --password-stdin <account>.dkr.ecr.<region>.amazonaws.com
docker tag my-fastq-pipeline:latest <account>.dkr.ecr.<region>.amazonaws.com/my-fastq-pipeline:latest
docker push <account>.dkr.ecr.<region>.amazonaws.com/my-fastq-pipeline:latest

부품 3: AWS Batch 잡 정의

Batch는 세 층의 개념입니다.

Compute environment: 실제 EC2 인스턴스를 관리하는 층. 최소 vCPU·최대 vCPU·인스턴스 타입 지정.

Job queue: 잡들이 대기하는 큐. 우선순위 지정.

Job definition: 어떤 컨테이너 이미지를 어떤 자원으로 어떻게 실행할지 정의.

python
import boto3


batch = boto3.client("batch")


batch.register_job_definition(
    jobDefinitionName="fastq-pipeline",
    type="container",
    containerProperties={
        "image": "<account>.dkr.ecr.<region>.amazonaws.com/my-fastq-pipeline:latest",
        "vcpus": 8,
        "memory": 32000,
        "command": [
            "--input", "Ref::input_s3",
            "--output", "Ref::output_s3"
        ],
        "jobRoleArn": "arn:aws:iam::<account>:role/BatchJobRole"
    }
)

Ref::input_s3 는 잡을 제출할 때 채워지는 파라미터입니다.

부품 4: 잡 제출

한 샘플에 대한 잡을 제출합니다.

python
def submit_sample_job(sample_id: str, input_s3: str, output_s3: str) -> str:
    response = batch.submit_job(
        jobName=f"fastq-{sample_id}",
        jobQueue="my-fastq-queue",
        jobDefinition="fastq-pipeline",
        parameters={
            "input_s3": input_s3,
            "output_s3": output_s3
        }
    )
    return response["jobId"]




job_id = submit_sample_job(
    sample_id="sample01",
    input_s3="s3://my-bucket/raw/sample01.fastq.gz",
    output_s3="s3://my-bucket/results/sample01/"
)
print(f"Submitted job: {job_id}")

네 부품을 합쳐 — 컨테이너 내부 로직

컨테이너 안에서 실행되는 pipeline.py 는 이렇게 생겼습니다.

python
import argparse
import subprocess
from pathlib import Path
import boto3




def parse_s3_uri(uri: str) -> tuple[str, str]:
    parts = uri.replace("s3://", "").split("/", 1)
    return parts[0], parts[1] if len(parts) > 1 else ""




def download_from_s3(s3_uri: str, local_path: Path) -> None:
    bucket, key = parse_s3_uri(s3_uri)
    s3 = boto3.client("s3")
    s3.download_file(bucket, key, str(local_path))




def upload_to_s3(local_path: Path, s3_uri: str) -> None:
    bucket, key = parse_s3_uri(s3_uri)
    s3 = boto3.client("s3")
    s3.upload_file(str(local_path), bucket, key)




def run_pipeline(input_s3: str, output_s3: str, work_dir: Path) -> None:
    work_dir.mkdir(exist_ok=True)
    
    raw = work_dir / "sample.fastq.gz"
    trimmed = work_dir / "trimmed.fastq.gz"
    aligned = work_dir / "aligned.bam"
    sorted_bam = work_dir / "sorted.bam"
    variants = work_dir / "variants.vcf"
    
    print(f"[1/5] Downloading {input_s3}")
    download_from_s3(input_s3, raw)
    
    print("[2/5] Quality trimming with fastp")
    subprocess.run(["fastp", "-i", str(raw), "-o", str(trimmed)], check=True)
    
    print("[3/5] Alignment with BWA")
    with open(aligned, "w") as f:
        subprocess.run(
            ["bwa", "mem", "/reference/hg38.fa", str(trimmed)],
            stdout=f, check=True
        )
    
    print("[4/5] Sort BAM")
    subprocess.run(
        ["samtools", "sort", str(aligned), "-o", str(sorted_bam)],
        check=True
    )
    
    print("[5/5] Variant calling")
    subprocess.run(
        ["bcftools", "call", "-o", str(variants), str(sorted_bam)],
        check=True
    )
    
    print(f"Uploading {variants} to {output_s3}variants.vcf")
    upload_to_s3(variants, f"{output_s3}variants.vcf")




if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True)
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    
    run_pipeline(args.input, args.output, Path("/tmp/work"))

이 스크립트가 여러분의 노트북에서도 돌아가고(작은 파일로), Batch 컨테이너에서도 돌아갑니다(큰 파일로). 차이는 파이썬 코드가 아닌 실행 환경입니다.

페이딩 — 여러분이 채워야 할 세 빈칸

빈칸 1: 대량 제출 오케스트레이터

한 샘플이 아닌 100개 샘플을 자동 제출합니다.

python
def submit_batch(sample_manifest: str, output_bucket: str) -> list[str]:
    """
    sample_manifest: CSV with columns [sample_id, input_s3]
    각 행에 대해 잡을 제출하고 job id 리스트 반환
    """
    import csv
    
    job_ids = []
    with open(sample_manifest) as f:
        reader = csv.DictReader(f)
        for row in reader:
            # TODO: submit_sample_job 호출
            # output_s3 를 output_bucket + sample_id 로 조합
            # 반환된 job_id 를 job_ids에 append
            pass
    
    return job_ids

힌트: output_s3 = f"s3://{output_bucket}/results/{row['sample_id']}/"

빈칸 2: 진행 상황 폴링

제출한 잡들의 상태를 주기적으로 확인합니다.

python
def wait_for_jobs(job_ids: list[str], poll_interval: int = 60) -> dict:
    """
    각 잡의 최종 상태를 반환.
    상태: SUBMITTED, PENDING, RUNNABLE, STARTING, RUNNING, SUCCEEDED, FAILED
    """
    import time
    
    final_states = {}
    pending = set(job_ids)
    
    while pending:
        # TODO: batch.describe_jobs(jobs=list(pending)) 호출
        # 각 잡 상태 확인, 완료된 것은 final_states에 기록하고 pending에서 제거
        # 남은 pending이 있으면 poll_interval 초 대기
        pass
    
    return final_states

힌트: response["jobs"] 각 항목의 status 필드를 확인. SUCCEEDED, FAILED만 최종 상태로 취급.

빈칸 3: 실패 재시도 로직

Spot 인스턴스는 저비용이지만 중간에 종료될 수 있습니다. 실패한 잡을 자동 재시도합니다.

python
def submit_with_retry(sample_id: str, input_s3: str, output_s3: str, max_attempts: int = 3) -> str:
    response = batch.submit_job(
        jobName=f"fastq-{sample_id}",
        jobQueue="my-fastq-queue",
        jobDefinition="fastq-pipeline",
        parameters={"input_s3": input_s3, "output_s3": output_s3},
        # TODO: retryStrategy 추가
        # attempts=max_attempts, evaluateOnExit 조건 지정
    )
    return response["jobId"]

힌트:

python
retryStrategy={
    "attempts": max_attempts,
    "evaluateOnExit": [
        {"onStatusReason": "Host EC2*", "action": "RETRY"},
        {"onExitCode": "0", "action": "EXIT"}
    ]
}

성찰 — 이 파이프라인이 실전 시퀀싱 파이프라인과 어떻게 다른가

워크플로 언어: 실전은 Nextflow, Snakemake, WDL 같은 워크플로 언어를 씁니다. 여러분의 파이썬 스크립트가 각 단계를 순차 실행하지만, 워크플로 언어는 DAG(Directed Acyclic Graph)로 스텝 의존성을 표현해 병렬 실행과 캐싱을 자동화합니다.

참조 데이터 관리: 여러분의 컨테이너에 hg38 참조가 하드코딩됐지만, 실전은 EFS(Elastic File System)나 별도 큰 볼륨에 참조를 두고 여러 잡이 공유합니다.

비용 최적화: 실전은 Spot 인스턴스 를 적극 활용해 비용을 70% 줄입니다. 그러나 Spot 종료 대응 로직이 필수입니다. 또한 각 스텝의 CPU/메모리 요구를 분리해 작은 스텝은 작은 인스턴스에서 돌립니다.

보안: 실전은 IAM role의 최소 권한 원칙, S3 버킷 암호화, VPC endpoint를 통한 프라이빗 네트워크, 감사 로그(CloudTrail) 등이 기본입니다.

관찰성: 각 잡의 stdout/stderr, 자원 사용률, 실행 시간을 CloudWatch로 수집. 실패 시 즉시 알림(SNS).

확장 프로젝트

1. Nextflow 이식: 위 파이썬 파이프라인을 Nextflow DSL2로 재작성. 각 스텝을 process로 분리해 병렬화 확인.

2. 결과 조회 대시보드: 완료된 잡의 결과 VCF를 자동 요약해 웹 대시보드로 표시. Streamlit이면 빠릅니다.

3. 비용 트래킹: 각 잡의 실제 사용 자원과 비용을 계산해 샘플당 비용 리포트 생성. Cost Explorer API 활용.

4. 로컬 개발 모드: LocalStack으로 로컬에서 S3/Batch를 시뮬레이션하는 개발 환경 구축. 실제 AWS 비용 없이 파이프라인 개발.

이 편의 부품 지도

  • [F] AWS 배포: Batch compute environment, job queue, job definition 설계.
  • [F] S3: 대량 데이터의 저장 + 접근. 리전·비용·전송의 이해.
  • [W] venv·file-io: 파이썬 가상환경과 로컬 파일 처리 (컨테이너 내부에서 사용).
  • [W] Docker: 컨테이너 이미지 빌드와 ECR push (완성 스크립트 제공).

[F] = 여러분이 직접 구현 / [W] = 완성 코드로 제공한 도구 개념.