Marat's Notes

AWS HealthOmics provides powerful tools for running genomic analysis pipelines at scale. This guide explores how to effectively use AWS HealthOmics for genomic data processing, with a focus on practical implementation, step-by-step instructions, and best practices.

Prerequisites

Before getting started, ensure you have:

AWS Account Setup:

# Install AWS CLI
curl "https://awscli.amazonaws.com/AWSCLIV2.pkg" -o "AWSCLIV2.pkg"
sudo installer -pkg AWSCLIV2.pkg -target /

# Configure AWS CLI
aws configure
# Enter your AWS Access Key ID
# Enter your AWS Secret Access Key
# Enter your default region (e.g., us-east-1)
# Enter your output format (json)

Required Tools:

# Install Python 3.8+
brew install python@3.8

# Create virtual environment
python3.8 -m venv healthomics-env
source healthomics-env/bin/activate

# Install required packages
pip install boto3 aws-healthomics-sdk pandas numpy

Initial Setup

1. AWS HealthOmics Configuration

# healthomics_config.py
import boto3
from aws_healthomics_sdk import HealthOmicsClient

class HealthOmicsConfig:
    def __init__(self):
        self.client = HealthOmicsClient(
            region_name='us-east-1',
            aws_access_key_id='YOUR_ACCESS_KEY',
            aws_secret_access_key='YOUR_SECRET_KEY'
        )
        
    def create_workflow(self, name, definition):
        """Create a new workflow"""
        try:
            response = self.client.create_workflow(
                name=name,
                definition=definition,
                description=f"Workflow for {name}"
            )
            return response['workflowId']
        except Exception as e:
            print(f"Error creating workflow: {e}")
            return None

    def list_workflows(self):
        """List all workflows"""
        try:
            response = self.client.list_workflows()
            return response['workflows']
        except Exception as e:
            print(f"Error listing workflows: {e}")
            return []

2. S3 Data Organization

# s3_manager.py
import boto3
from pathlib import Path

class S3Manager:
    def __init__(self):
        self.s3 = boto3.client('s3')
        self.bucket = 'your-genomics-bucket'
        
    def upload_reference_genome(self, local_path, s3_key):
        """Upload reference genome to S3"""
        try:
            self.s3.upload_file(
                local_path,
                self.bucket,
                f"reference_genomes/{s3_key}"
            )
            return f"s3://{self.bucket}/reference_genomes/{s3_key}"
        except Exception as e:
            print(f"Error uploading reference genome: {e}")
            return None

    def upload_sequencing_data(self, local_path, s3_key):
        """Upload sequencing data to S3"""
        try:
            self.s3.upload_file(
                local_path,
                self.bucket,
                f"sequencing_data/{s3_key}"
            )
            return f"s3://{self.bucket}/sequencing_data/{s3_key}"
        except Exception as e:
            print(f"Error uploading sequencing data: {e}")
            return None

Pipeline Implementation

1. Workflow Definition

# workflow_definition.py
class WorkflowDefinition:
    def __init__(self):
        self.definition = {
            "name": "Genomic Analysis Pipeline",
            "version": "1.0",
            "steps": [
                {
                    "name": "Quality Control",
                    "tool": "FastQC",
                    "inputs": {
                        "reads": "${input.reads}"
                    },
                    "outputs": {
                        "report": "qc_report.html"
                    }
                },
                {
                    "name": "Alignment",
                    "tool": "BWA",
                    "inputs": {
                        "reads": "${input.reads}",
                        "reference": "${input.reference}"
                    },
                    "outputs": {
                        "bam": "aligned.bam"
                    }
                },
                {
                    "name": "Variant Calling",
                    "tool": "GATK",
                    "inputs": {
                        "bam": "${steps.alignment.outputs.bam}",
                        "reference": "${input.reference}"
                    },
                    "outputs": {
                        "vcf": "variants.vcf"
                    }
                }
            ]
        }

    def get_definition(self):
        return self.definition

2. Pipeline Execution

# pipeline_executor.py
from healthomics_config import HealthOmicsConfig
from s3_manager import S3Manager

class PipelineExecutor:
    def __init__(self):
        self.healthomics = HealthOmicsConfig()
        self.s3 = S3Manager()
        
    def run_pipeline(self, workflow_id, input_data):
        """Run a genomic pipeline"""
        try:
            response = self.healthomics.client.start_run(
                workflowId=workflow_id,
                input=input_data,
                name=f"Run_{workflow_id}_{input_data['sample_id']}"
            )
            return response['runId']
        except Exception as e:
            print(f"Error running pipeline: {e}")
            return None

    def monitor_run(self, run_id):
        """Monitor pipeline execution"""
        try:
            response = self.healthomics.client.get_run(runId=run_id)
            return {
                'status': response['status'],
                'progress': response['progress'],
                'startTime': response['startTime'],
                'endTime': response.get('endTime')
            }
        except Exception as e:
            print(f"Error monitoring run: {e}")
            return None

Data Management

1. Input Data Preparation

# data_preparation.py
import pandas as pd
from pathlib import Path

class DataPreparation:
    def __init__(self):
        self.metadata = pd.DataFrame()
        
    def prepare_sequencing_data(self, data_dir):
        """Prepare sequencing data for pipeline"""
        try:
            # Read metadata
            self.metadata = pd.read_csv(data_dir / 'metadata.csv')
            
            # Validate data
            for _, row in self.metadata.iterrows():
                self._validate_sample(row)
                
            return True
        except Exception as e:
            print(f"Error preparing data: {e}")
            return False
            
    def _validate_sample(self, sample):
        """Validate individual sample"""
        required_files = ['R1.fastq.gz', 'R2.fastq.gz']
        sample_dir = Path(sample['sample_dir'])
        
        for file in required_files:
            if not (sample_dir / file).exists():
                raise ValueError(f"Missing required file: {file}")

2. Output Management

# output_manager.py
import json
from datetime import datetime

class OutputManager:
    def __init__(self, s3_manager):
        self.s3 = s3_manager
        
    def organize_outputs(self, run_id, outputs):
        """Organize pipeline outputs"""
        try:
            # Create output structure
            output_structure = {
                'run_id': run_id,
                'timestamp': datetime.now().isoformat(),
                'outputs': outputs
            }
            
            # Save to S3
            self.s3.upload_json(
                output_structure,
                f"pipeline_outputs/{run_id}/summary.json"
            )
            
            return True
        except Exception as e:
            print(f"Error organizing outputs: {e}")
            return False

Quality Control

1. Data Quality Checks

# quality_control.py
import subprocess
from pathlib import Path

class QualityControl:
    def __init__(self):
        self.tools = {
            'fastqc': 'fastqc',
            'multiqc': 'multiqc'
        }
        
    def run_quality_checks(self, input_dir, output_dir):
        """Run quality control checks"""
        try:
            # Run FastQC
            subprocess.run([
                self.tools['fastqc'],
                str(input_dir),
                '-o', str(output_dir)
            ], check=True)
            
            # Run MultiQC
            subprocess.run([
                self.tools['multiqc'],
                str(output_dir),
                '-o', str(output_dir)
            ], check=True)
            
            return True
        except Exception as e:
            print(f"Error running quality checks: {e}")
            return False

2. Pipeline Validation

# pipeline_validator.py
class PipelineValidator:
    def __init__(self):
        self.required_outputs = {
            'alignment': ['bam', 'bai'],
            'variant_calling': ['vcf', 'tbi']
        }
        
    def validate_pipeline_outputs(self, run_id, outputs):
        """Validate pipeline outputs"""
        try:
            for step, required_files in self.required_outputs.items():
                if step not in outputs:
                    raise ValueError(f"Missing step: {step}")
                    
                for file in required_files:
                    if file not in outputs[step]:
                        raise ValueError(f"Missing output file: {file}")
                        
            return True
        except Exception as e:
            print(f"Error validating outputs: {e}")
            return False

Cost Management

1. Resource Optimization

# resource_optimizer.py
class ResourceOptimizer:
    def __init__(self):
        self.instance_types = {
            'small': {'vcpu': 2, 'memory': 4},
            'medium': {'vcpu': 4, 'memory': 8},
            'large': {'vcpu': 8, 'memory': 16}
        }
        
    def optimize_resources(self, input_size, complexity):
        """Optimize resource allocation"""
        try:
            if input_size < 10 and complexity == 'low':
                return self.instance_types['small']
            elif input_size < 50 and complexity == 'medium':
                return self.instance_types['medium']
            else:
                return self.instance_types['large']
        except Exception as e:
            print(f"Error optimizing resources: {e}")
            return self.instance_types['medium']

2. Cost Tracking

# cost_tracker.py
import boto3
from datetime import datetime, timedelta

class CostTracker:
    def __init__(self):
        self.ce = boto3.client('ce')
        
    def track_pipeline_costs(self, run_id, start_time, end_time):
        """Track pipeline costs"""
        try:
            response = self.ce.get_cost_and_usage(
                TimePeriod={
                    'Start': start_time.isoformat(),
                    'End': end_time.isoformat()
                },
                Granularity='DAILY',
                Filter={
                    'Tags': {
                        'Key': 'RunId',
                        'Values': [run_id]
                    }
                }
            )
            
            return response['ResultsByTime']
        except Exception as e:
            print(f"Error tracking costs: {e}")
            return []

Best Practices

1. Pipeline Design

# pipeline_design.py
class PipelineDesign:
    def __init__(self):
        self.best_practices = {
            'modularity': True,
            'error_handling': True,
            'logging': True,
            'checkpointing': True
        }
        
    def implement_best_practices(self, workflow_definition):
        """Implement best practices in workflow"""
        try:
            # Add error handling
            for step in workflow_definition['steps']:
                step['error_handling'] = {
                    'retry_count': 3,
                    'retry_delay': 300
                }
                
            # Add logging
            workflow_definition['logging'] = {
                'level': 'INFO',
                'destination': 's3://your-bucket/logs'
            }
            
            # Add checkpointing
            workflow_definition['checkpointing'] = {
                'enabled': True,
                'interval': 3600
            }
            
            return workflow_definition
        except Exception as e:
            print(f"Error implementing best practices: {e}")
            return workflow_definition

2. Execution Management

# execution_manager.py
class ExecutionManager:
    def __init__(self):
        self.max_concurrent_runs = 5
        self.active_runs = set()
        
    def manage_execution(self, run_id, workflow_id, input_data):
        """Manage pipeline execution"""
        try:
            # Check concurrent runs
            if len(self.active_runs) >= self.max_concurrent_runs:
                raise Exception("Maximum concurrent runs reached")
                
            # Start run
            self.active_runs.add(run_id)
            
            # Monitor run
            while True:
                status = self.monitor_run(run_id)
                if status['status'] in ['COMPLETED', 'FAILED']:
                    self.active_runs.remove(run_id)
                    break
                    
            return status
        except Exception as e:
            print(f"Error managing execution: {e}")
            return None

Conclusion

Running genomic pipelines with AWS HealthOmics requires careful planning and implementation. By following this guide, you can:

Set up efficient pipelines
Optimize resource usage
Ensure data quality
Maintain security
Control costs

Remember to:

Regularly review pipeline performance
Optimize resource allocation
Monitor costs
Update security measures
Follow best practices

With proper implementation and maintenance, AWS HealthOmics can provide powerful tools for running genomic analysis pipelines at scale while maintaining efficiency and security.

Running Genomic Pipelines with AWS HealthOmics