DataTurbo:A high-performance LLM data processing pipeline framework

---

✨ Features

• 🚀 Queue-based Streaming Architecture: Efficient data flow between pipeline stages with non-blocking I/O
• 💻 Heterogeneous Computing Support: Hybrid CPU/GPU execution with fine-grained resource allocation
• 📊 AIMD Dynamic Batch Sizing: Adaptive batch size tuning using Additive Increase Multiplicative Decrease algorithm
• 📝 Recipe-driven Configuration: Declarative YAML/JSON pipeline definitions for easy customization
• 🔧 Extensible Operator System: Simple base class for creating custom operators
• ⚖️ Load-balanced Scheduling: Intelligent worker selection for optimal throughput
• 🛡️ Fault Tolerance: Two-layer checkpointing with automatic recovery and health monitoring
• 🎯 Bayesian Hyperparameter Optimization: Optional multi-fidelity tuning for performance optimization

📦 Installation

# Clone the repository
git clone https://github.com/yourusername/DataTurbo.git
cd DataTurbo

# Install in development mode (recommended for trying examples)
pip install -e .

# Or install from PyPI (coming soon)
# pip install dataturbo

Requirements:

• Python 3.8+
• Ray 2.0+
• 8GB+ RAM recommended
• CUDA-compatible GPU (optional, for GPU operators)

Reproducibility (GitHub Action)

We provide a GitHub Action workflow for reproducible experiments. It runs the math pipeline demo on each push to main or on manual trigger (workflow_dispatch), so reviewers can verify the pipeline without a local setup.

• Workflow: .github/workflows/reproduce.yml — installs dependencies and runs the demo.
• Demo script: scripts/reproduce.sh — invokes the math pipeline with demo data (e.g. --input ./demo_data, --output ./output/math_alpha_test, --max-size 30, --enable-ft).

You can edit scripts/reproduce.sh to change the launch configuration (recipe, input path, output dir, --max-size, --enable-ft, etc.).

Note: The demo uses a shared API with rate limits. For runs with more than ~30 items, replace the API key and base URL in your recipe (e.g. examples/math_pipeline/recipe.yaml) with your own endpoint to avoid throttling.

🚀 Quick Start

The easiest way to get started is to run one of our example pipelines:

# 1. Install DataTurbo and requirements.txt
cd DataTurbo
pip install -e .
pip install -r requirements.txt

# 2. Navigate to math pipeline example
cd examples/math_pipeline

# 3. Run the pipeline with your data
./run.sh \
    --recipe recipe.yaml \
    --input /path/to/your/data/ \
    --output output/math_processed \
    --max-size 300 \
    # --enable-ft

What this does:

• ✅ Loads your math problem dataset
• ✅ Runs quality assessment, domain classification, and difficulty analysis
• ✅ Outputs processed results to output/processed/final_output.json
• ✅ Generates performance metrics in output/processed/metrics.json

Available Examples:

# Math dataset processing
cd examples/math_pipeline && ./run.sh --input data.json --output output/

# Code dataset processing  
cd examples/code_pipeline && ./run.sh --input code.json --output output/

# Multimodal (image-text) processing
cd examples/multimodal_pipeline && ./run.sh --input images.json --output output/

---

📚 Usage

DataTurbo provides two ways to build and run data processing pipelines:

Method 1: Recipe-driven Configuration (Recommended)

Use declarative YAML/JSON files to define your pipeline. This approach separates configuration from code and makes it easy to modify pipelines without changing code.

Option A: Using Bash Script

cd examples/math_pipeline

./run.sh \
    --recipe recipe.yaml \
    --input /path/to/your/data/ \
    --output output/math_processed \
    --max-size 1000 \
    --enable-ft

Script Parameters:

Parameter	Required	Default	Description
--recipe	No	recipe.yaml	Recipe configuration file
--input	Yes	-	Input data path (file or directory)
--output	Yes	-	Output directory
--max-size	No	All data	Maximum number of items to process
--enable-ft	No	Disabled	Enable fault tolerance (checkpointing & recovery)

Customization: Each example includes a recipe.yaml file you can modify to customize the pipeline:

• Change batch sizes and worker counts
• Add/remove processing operators
• Adjust model parameters and API settings
• Configure CPU/GPU resource allocation

Option B: Using Python with RecipeParser

from dataturbo import RecipeParser, OperatorRegistry, QueueBasedPipeline

# Create operator registry
registry = OperatorRegistry()
# Register your operators here...

# Load recipe
parser = RecipeParser(registry)
config, stage_configs = parser.create_pipeline_from_recipe(
    recipe_path="pipeline.yaml",
    input_path="data/input.json",
    output_dir="output/"
)

# Run pipeline
pipeline = QueueBasedPipeline(config, stage_configs)
metrics = pipeline.run("data/input.json", "output/")

print(f"Processed {metrics['input_count']} items in {metrics['total_time']:.2f}s")

When to use Recipe-driven:

• ✅ Quick prototyping and experimentation
• ✅ Production deployments with standardized configs
• ✅ When non-developers need to adjust pipeline settings
• ✅ When you want configuration version control separate from code

---

Method 2: Python API (Programmatic)

Define your pipeline entirely in Python code for maximum flexibility and programmatic control.

import ray
import sys
from pathlib import Path

# Add examples to path to import custom operators
sys.path.insert(0, str(Path(__file__).parent / 'examples' / 'math_pipeline'))

from dataturbo import QueueBasedPipeline, PipelineConfig, ModelConfig
from math_pipeline.operators import QualityAssessmentActor, DomainClassificationActor
from math_pipeline.config import QualityAssessmentConfig, DomainClassificationConfig

# Initialize Ray
ray.init()

# Create configuration
model_config = ModelConfig(
    name="Qwen2.5-0.5B-Instruct",
    api_key="your-api-key",
    base_url="http://localhost:8000/v1"
)

config = PipelineConfig(
    input_path="data/input.json",
    output_dir="output/",
    model=model_config,
    pipeline_batch_size=50,
    workers_per_stage=2
)

# Create operator configurations
quality_config = QualityAssessmentConfig(batch_size=50, workers_per_stage=2)
domain_config = DomainClassificationConfig(batch_size=50, workers_per_stage=2)

# Define stage configurations
stage_configs = {
    'quality': (QualityAssessmentActor, quality_config, model_config),
    'domain': (DomainClassificationActor, domain_config, model_config),
}

# Create and run pipeline
pipeline = QueueBasedPipeline(config, stage_configs)
metrics = pipeline.run("data/input.json", "output/")

print(f"Processed {metrics['input_count']} items in {metrics['total_time']:.2f}s")
print(f"Throughput: {metrics['throughput']:.2f} items/s")

When to use Python API:

• ✅ Embedding DataTurbo into larger Python applications
• ✅ Dynamic pipeline construction based on runtime conditions
• ✅ Fine-grained control over every aspect of the pipeline
• ✅ Programmatic processing of pipeline results

Note: DataTurbo provides the framework and BaseOperator base class. All operator implementations are in examples/ as references. Copy and modify them for your needs.

---

Project Structure

DataTurbo/
├── .github/workflows/         # GitHub Actions (reproducibility)
│   └── reproduce.yml
├── assets/                    # Images for README/docs
├── dataturbo/                 # Core framework package
│   ├── core/                  # Pipeline engine (QueueBasedPipeline + StageCoordinator)
│   │   ├── pipeline.py
│   │   ├── coordinator.py
│   │   └── base_operator.py
│   ├── recipe/                # Recipe parser + operator registry
│   ├── config/                # Pipeline/Model configs
│   ├── fault_tolerance/       # Optional checkpointing + monitoring
│   ├── operators/             # Optional place for user-defined operators (empty by default)
│   └── utils/                 # Utilities
├── examples/                  # Example pipelines (operators live here)
│   ├── math_pipeline/
│   │   ├── operators.py
│   │   ├── config.py
│   │   ├── recipe.yaml
│   │   ├── run_from_recipe.py
│   │   ├── run.sh
│   │   └── demo_data/
│   ├── code_pipeline/
│   │   └── ...
│   └── multimodal_pipeline/
│       └── ...
├── scripts/
│   └── reproduce.sh           # Local reproducibility entrypoint (used by GitHub Action)
├── docs/                      # Documentation
├── optimization/              # Bayesian optimization module
├── LICENSE
├── README.md
├── requirements.txt
├── setup.py
└── pyproject.toml

Creating Custom Operators

Extend BaseOperator to create custom operators:

import ray
from dataturbo.core.base_operator import BaseOperator

@ray.remote
class MyCustomOperator(BaseOperator):
    def __init__(self, config, model_config=None, gpu_id=None):
        super().__init__(config, model_config, gpu_id)
        # Initialize your operator
    
    def process_chunk(self, chunk):
        # Process data and return results
        results = {}
        for item in chunk:
            # Your processing logic
            results[item['id']] = processed_value
        return {'results': results}

Recipe Format

name: "My Pipeline"
version: "1.0"

global:
  model_name: "your-model"
  api_key: "your-api-key"
  base_url: "http://localhost:8000/v1"
  pipeline_batch_size: 100

operators:
  - name: deduplication
    type: deduplication
    params:
      similarity_threshold: 0.85
      workers_per_stage: 2
  
  - name: quality_check
    type: quality_assessment
    params:
      batch_size: 50
      device: gpu
      num_gpus: 0.5

Hyperparameter Optimization

Use the optimization module for automatic hyperparameter tuning:

from optimization import MultiFidelityTuner, FidelityLevel

tuner = MultiFidelityTuner(
    objective_fn=my_objective,
    param_space={
        'batch_size': (50, 200, 10),
        'workers': [2, 3, 4, 5]
    },
    fidelity_levels=[
        FidelityLevel("quick", 100, 30),
        FidelityLevel("final", 1000, 15)
    ]
)

result = tuner.optimize()

License

Apache License 2.0

Contributing

Contributions are welcome! Please read our contributing guidelines before submitting PRs.