Data Pipeline Manager

Purpose

The Data Pipeline Manager skill provides comprehensive guidance for designing, implementing, and troubleshooting data processing pipelines with emphasis on quality validation, error handling, and reliability. Data pipelines are a critical component of modern bioinformatics and data science workflows, yet they are a prevalent source of failures in automated systems. Poor data quality, format mismatches, missing files, and transient errors can cascade through pipelines, causing failures that are difficult to diagnose and fix.

This skill addresses these challenges by providing structured approaches to pipeline design, validation strategies at each stage, robust error handling patterns, and monitoring techniques. Whether you're building a new RNA-seq analysis pipeline, debugging a failed ETL job, or ensuring data quality across complex multi-stage workflows, this skill provides the frameworks and best practices needed for success.

The skill integrates closely with the bioinformatician skill (which implements specific bioinformatics pipelines) and the systems-architect skill (which designs the overall system architecture), providing the critical middle layer of pipeline orchestration, validation, and reliability.

When to Use This Skill

Use the Data Pipeline Manager skill when you need to:

Pipeline Design and Implementation

Debugging and Troubleshooting

Quality Assurance

Error Recovery and Reliability

Pipeline Optimization

Core Workflow

The Data Pipeline Manager follows a structured six-stage workflow that ensures robust, reliable pipeline operation:

1. Design Phase

**Define Pipeline Architecture**

Begin by clearly defining the pipeline's purpose, inputs, outputs, and transformations. Use a structured approach:

**Choose Pipeline Pattern**

Select the appropriate pipeline pattern based on your requirements:

**Design for Observability**

Build in observability from the start:

2. Input Validation

**Pre-Flight Checks**

Before processing begins, validate all inputs thoroughly:

**Content Validation**

Beyond basic format checks, validate data content:

**Early Failure Strategy**

Fail fast on input validation errors rather than processing invalid data:

3. Transform Phase

**Stage-by-Stage Processing**

Execute each transformation stage with careful attention to reliability:

**Transformation Best Practices**

Follow these practices for robust transformations:

**Handling Large Data**

For large-scale data processing:

4. Output Validation

**Structure Validation**

Verify output structure meets specifications:

**Content Validation**

Validate output data quality and correctness:

**Quality Metrics**

Calculate and report quality metrics:

5. Error Handling

**Error Classification**

Classify errors to determine appropriate handling:

**Retry Logic**

Implement intelligent retry strategies:

**Logging and Diagnostics**

Create comprehensive error logs:

**Checkpointing for Resumability**

Enable pipeline resumption after failures:

6. Monitoring and Alerting

**Real-Time Monitoring**

Track pipeline health in real-time:

**Alerting Strategy**

Set up intelligent alerts:

**Dashboard Design**

Create effective monitoring dashboards:

Pipeline Patterns

Sequential Pipeline

**Description**: Each stage depends on the previous stage's output. Data flows linearly through the pipeline.

**Use Cases**:

**Advantages**:

**Disadvantages**:

**Implementation Considerations**:

Parallel Pipeline

**Description**: Multiple independent branches process data simultaneously without dependencies.

**Use Cases**:

**Advantages**:

**Disadvantages**:

**Implementation Considerations**:

Conditional Pipeline

**Description**: Pipeline branches based on data content or intermediate results.

**Use Cases**:

**Advantages**:

**Disadvantages**:

**Implementation Considerations**:

Hybrid Pipeline

**Description**: Combination of sequential, parallel, and conditional patterns for complex workflows.

**Use Cases**:

**Advantages**:

**Disadvantages**:

**Implementation Considerations**:

Quality Validation Strategies

Input Validation

**File Existence and Accessibility**

```python

Check file exists and is readable

if not os.path.exists(input_file):

raise FileNotFoundError(f"Input file not found: {input_file}")

if not os.access(input_file, os.R_OK):

raise PermissionError(f"Cannot read input file: {input_file}")

```

**Format Validation**

```python

Validate FASTQ format

def validate_fastq(file_path):

with open(file_path) as f:

line_count = 0

for line in f:

line_count += 1

if line_count % 4 == 1 and not line.startswith('@'):

raise ValueError(f"Invalid FASTQ format at line {line_count}")

if line_count % 4 == 3 and not line.startswith('+'):

raise ValueError(f"Invalid FASTQ format at line {line_count}")

```

**Non-Empty Validation**

```python

Check file is not empty

file_size = os.path.getsize(input_file)

if file_size == 0:

raise ValueError(f"Input file is empty: {input_file}")

if file_size < expected_min_size:

raise ValueError(f"Input file is too small: {file_size} bytes")

```

**Schema Validation**

```python

Validate CSV schema

required_columns = ['sample_id', 'condition', 'batch']

df = pd.read_csv(metadata_file)

missing_cols = set(required_columns) - set(df.columns)

if missing_cols:

raise ValueError(f"Missing required columns: {missing_cols}")

```

Intermediate Validation

**Dimension Checking**

```python

Validate array dimensions

if counts_matrix.shape[0] != n_genes:

raise ValueError(f"Expected {n_genes} genes, got {counts_matrix.shape[0]}")

if counts_matrix.shape[1] != n_samples:

raise ValueError(f"Expected {n_samples} samples, got {counts_matrix.shape[1]}")

```

**Missing Data Detection**

```python

Check for NaN and Inf values

if np.isnan(data).any():

nan_count = np.isnan(data).sum()

raise ValueError(f"Data contains {nan_count} NaN values")

if np.isinf(data).any():

inf_count = np.isinf(data).sum()

raise ValueError(f"Data contains {inf_count} Inf values")

```

**Range Validation**

```python

Validate numeric ranges

if (data < 0).any():

raise ValueError("Data contains negative values (expected non-negative)")

if data.max() > expected_max:

raise ValueError(f"Data contains values above {expected_max}")

```

Output Validation

**Completeness Checking**

```python

Verify all expected outputs exist

expected_outputs = ['counts.csv', 'qc_metrics.json', 'results.html']

for output_file in expected_outputs:

if not os.path.exists(output_file):

raise ValueError(f"Missing expected output: {output_file}")

```

**Distribution Validation**

```python

Check output distributions are reasonable

mean_count = counts_matrix.mean()

if mean_count < 100:

warnings.warn(f"Low mean count: {mean_count} (expected > 100)")

if counts_matrix.std() / mean_count > 10:

warnings.warn(f"High coefficient of variation: {counts_matrix.std() / mean_count}")

```

**Sanity Checks**

```python

Domain-specific sanity checks

if alignment_rate < 0.5:

warnings.warn(f"Low alignment rate: {alignment_rate:.1%} (expected > 50%)")

if duplication_rate > 0.3:

warnings.warn(f"High duplication rate: {duplication_rate:.1%} (expected < 30%)")

```

Error Handling Strategies

Transient Error Retry

**Exponential Backoff**

```python

import time

from functools import wraps

def retry_with_backoff(max_retries=3, base_delay=1):

def decorator(func):

@wraps(func)

def wrapper(*args, **kwargs):

for attempt in range(max_retries):

try:

return func(*args, **kwargs)

except TransientError as e:

if attempt == max_retries - 1:

raise

delay = base_delay * (2 ** attempt)

time.sleep(delay)

return None

return wrapper

return decorator

```

Checkpoint Implementation

**Basic Checkpointing**

```python

def process_with_checkpoint(items, checkpoint_file, process_func):

# Load checkpoint if exists

completed = set()

if os.path.exists(checkpoint_file):

with open(checkpoint_file) as f:

completed = set(json.load(f))

# Process items, skipping completed ones

for item in items:

if item in completed:

continue

process_func(item)

completed.add(item)

# Save checkpoint

with open(checkpoint_file, 'w') as f:

json.dump(list(completed), f)

```

Comprehensive Error Logging

**Structured Error Context**

```python

import logging

import json

def log_pipeline_error(stage, error, context):

error_data = {

'timestamp': datetime.now().isoformat(),

'stage': stage,

'error_type': type(error).__name__,

'error_message': str(error),

'context': context,

'traceback': traceback.format_exc()

}

logging.error(json.dumps(error_data))

```

Bioinformatics-Specific Considerations

FASTQ to Count Matrix Pipeline

**Standard Workflow**:

1. **Quality Control**: FastQC on raw FASTQ files

2. **Preprocessing**: Trimming adapters and low-quality bases (Trimmomatic, cutadapt)

3. **Alignment**: Align to reference genome (STAR, HISAT2, BWA)

4. **Quality Control**: Check alignment metrics (samtools, picard)

5. **Quantification**: Count reads per gene (featureCounts, HTSeq, RSEM)

6. **Count Matrix**: Assemble counts into matrix for downstream analysis

**Critical Validation Points**:

Genome Build Consistency

**Problem**: Mixing genome builds (hg19 vs hg38, mm9 vs mm10) causes misalignments and incorrect results.

**Solution**:

Sample Metadata Tracking

**Essential Metadata**:

**Validation**:

Reference Data Management

**Best Practices**:

Integration with Other Skills

Integration with Bioinformatician Skill

The Data Pipeline Manager provides the framework and infrastructure for pipelines that the bioinformatician skill implements:

Integration with Systems Architect Skill

The Data Pipeline Manager implements pipelines within infrastructure designed by systems-architect:

Summary

The Data Pipeline Manager skill provides comprehensive guidance for building robust, reliable data processing pipelines. By following the structured workflow of design, input validation, transformation, output validation, error handling, and monitoring, you can create pipelines that handle errors gracefully, validate data quality at every stage, and provide clear diagnostics when issues occur.

Key principles:

Whether you're building new pipelines or troubleshooting existing ones, this skill provides the patterns, strategies, and best practices needed for success in bioinformatics and data processing workflows.