FASTQ Cloud Batch – Moving Out of Notebooks and Onto AWS

After completing this topic

You will be able to create your own pipeline that wraps your local Python script in a container and runs it in bulk using AWS Batch, combining the AWS deployment and S3 storage concepts learned in the textbook. Although FASTQ files are used as an example, this pattern can be applied to any type of bulk data processing.

This article is a general educational example. Actual production sequencing pipelines use workflow languages such as Nextflow, Snakemake, or WDL.

"My Laptop's Disk Is Full" — The Limitations of Local Processing

Let's say you've received a 200GB FASTQ file from a sequencing core. You want to perform the following operations on this file:

1. Quality trimming (removing low-quality reads)
2. Alignment to a reference genome
3. Variant calling

Local approach:

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"])

This approach has three practical problems:

Problem 1: Disk space. The original 200GB + multiple intermediate files = over 1TB. Your laptop's disk likely doesn't have that much space.

Problem 2: Memory and time. BWA alignment requires 32GB of memory and takes several hours. Your laptop can't perform other tasks during this time.

Problem 3: Scalability. This time it's just one sample, but the next project will have 100 samples. Local sequential processing will take weeks.

The real approach is to wrap this pipeline in a container and run it on AWS Batch. The original file is stored in S3, and the Batch job spins up a container on a large EC2 instance to process the file, and then stores the results back in S3. 100 samples will run as 100 parallel jobs.

From Black Box to Components: Unveiling a Cloud Pipeline

There are four key components.

Component 1: S3 – Infinite Storage

S3 is essentially infinite object storage. It allows for files up to 5TB in size, with no limit on the number of files per bucket. Lifecycle policies can be used to automatically move older files to lower-cost tiers (Glacier).

Access it with the Python SDK (boto3).

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"])

Note: S3 costs include storage + transfer + requests. If you access it from an EC2 instance in the same region, there are no transfer costs. The first rule of cost-effectiveness is to run it in the region where your files are located.

Component 2: Docker Container

Package everything your pipeline script needs – Python + fastp + bwa + samtools + bcftools + your code – into a single image.

Example 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"]

Build and push to ECR (Elastic Container Registry):

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

Component 3: AWS Batch Job Definition

Batch has three layers of concepts.

Compute environment: The layer that manages the actual EC2 instances. Specify minimum vCPUs, maximum vCPUs, and instance type.

Job queue: The queue where jobs wait. Priority can be assigned.

Job definition: Defines which container image to run, with what resources, and how.

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 is a parameter that will be filled in when the job is submitted.

Component 4: Job Submission

Submit a job for a single sample.

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}")

Putting It All Together: The Container's Internal Logic

The pipeline.py script that runs inside the container looks like this:

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"))

This script works both on your laptop (with a small file) and in the Batch container (with a large file). The difference is not in the Python code, but in the execution environment.

Fading — Three Blanks for You to Fill

Blank 1: Batch Submission Orchestrator

Automatically submit 100 samples instead of just one.

def submit_batch(sample_manifest: str, output_bucket: str) -> list[str]:
    """
    sample_manifest: CSV with columns [sample_id, input_s3]
    For each row, submit a job and return a list of job IDs.
    """
    import csv
    
    job_ids = []
    with open(sample_manifest) as f:
        reader = csv.DictReader(f)
        for row in reader:
            # TODO: Call submit_sample_job
            # Combine output_s3 with output_bucket + sample_id
            # Append the returned job_id to job_ids
            pass
    
    return job_ids

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

Blank 2: Progress Polling

Periodically check the status of submitted jobs.

def wait_for_jobs(job_ids: list[str], poll_interval: int = 60) -> dict:
    """
    Return the final status of each job.
    Statuses: SUBMITTED, PENDING, RUNNABLE, STARTING, RUNNING, SUCCEEDED, FAILED
    """
    import time
    
    final_states = {}
    pending = set(job_ids)
    
    while pending:
        # TODO: Call batch.describe_jobs(jobs=list(pending))
        # Check the status of each job, record completed jobs in final_states, and remove them from pending.
        # If there are remaining pending jobs, wait for poll_interval seconds.
        pass
    
    return final_states

Hint: Check the status field of each item in response["jobs"]. Treat only SUCCEEDED and FAILED as final statuses.

Blank 3: Failure Retry Logic

Spot instances are low-cost but can be terminated mid-process. Automatically retry failed jobs.

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: Add retryStrategy
        # Specify attempts and evaluateOnExit condition
    )
    return response["jobId"]

Hint:

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

Reflection — How does this pipeline differ from a production sequencing pipeline?

Workflow Language: Production pipelines use workflow languages such as Nextflow, Snakemake, or WDL. While your Python script sequentially executes each step, workflow languages express step dependencies as a DAG (Directed Acyclic Graph), which automates parallel execution and caching.

Reference Data Management: Your container has the hg38 reference hardcoded, but in production, references are stored on EFS (Elastic File System) or a separate large volume and shared across multiple jobs.

Cost Optimization: Production pipelines actively leverage Spot Instances to reduce costs by 70%. However, logic to handle Spot Instance terminations is essential. Additionally, the CPU/memory requirements for each step are separated, allowing smaller steps to run on smaller instances.

Security: In production, the principle of least privilege for IAM roles, S3 bucket encryption, a private network via VPC endpoints, and audit logs (CloudTrail) are standard.

Observability: Each job's stdout/stderr, resource utilization, and execution time are collected in CloudWatch. Notifications are sent immediately upon failure (SNS).

Expansion Project

1. Nextflow Porting: Rewrite the above Python pipeline using Nextflow DSL2. Separate each step into a process and verify parallelization.

2. Result Viewing Dashboard: Automatically summarize the VCF results of completed jobs and display them on a web dashboard. Streamlit would be fast.

3. Cost Tracking: Calculate the actual resources used and the cost for each job, and generate a cost report per sample. Utilize the Cost Explorer API.

4. Local Development Mode: Build a development environment that simulates S3/Batch locally using LocalStack. Develop the pipeline without incurring actual AWS costs.

Component Guide for This Tutorial

• [F] AWS Deployment: Designing the batch compute environment, job queue, and job definition.
• [F] S3: Storing and accessing large amounts of data. Understanding region, cost, and transfer.
• [W] venv & file-io: Python virtual environment and local file handling (used within the container).
• [W] Docker: Building container images and pushing to ECR (a complete script will be provided).

[F] = Concepts you will implement yourself / [W] = Tool concepts provided with complete code.