dataghost 0.1.1

Time-Travel Debugger for Data Pipelines

DataGhost 👻

Time-Travel Debugger for Data Pipelines

Debug your data pipelines like you debug your code. DataGhost captures complete execution snapshots, enables precise replay of historical runs, and provides rich comparison tools - all with zero configuration.

---

🚀 Quick Start

Installation

# Basic installation
pip install dataghost

# With dashboard support
pip install dataghost[dashboard]

# For Google Colab/Jupyter
pip install dataghost[colab]

30-Second Demo

from ttd import snapshot

@snapshot(task_id="my_pipeline")
def process_data(data: list, multiplier: int = 2) -> dict:
    processed = [x * multiplier for x in data]
    return {
        "processed_data": processed,
        "count": len(processed),
        "average": sum(processed) / len(processed)
    }

# Run your function normally - snapshots are captured automatically
result = process_data([1, 2, 3, 4, 5], multiplier=3)

View Your Data

# See all your pipeline runs
dataghost overview

# Launch interactive dashboard
dataghost dashboard

# For Google Colab (creates public URL)
dataghost dashboard --tunnel

---

✨ Key Features

🎯 Zero-Config Snapshots

Just add @snapshot decorator - no setup required. Captures inputs, outputs, metadata, and execution context automatically.

🔄 Time-Travel Debugging

Replay any historical execution with identical conditions. Perfect for debugging failures and testing changes.

📊 Rich Comparisons

Compare runs side-by-side with structured diffing. See exactly what changed between executions.

🌐 Cloud-Ready

Works seamlessly in Google Colab, Jupyter notebooks, and remote environments with automatic tunnel support.

📱 Beautiful Dashboard

Interactive web interface with real-time monitoring, performance analytics, and one-click operations.

🏗️ Framework Integration

First-class support for Apache Airflow, with more integrations coming soon.

---

🎮 Interactive Demo

Try DataGhost in your browser:

---

📖 Complete Guide

1. Basic Usage

Capture Snapshots

from ttd import snapshot

@snapshot(task_id="data_processing")
def transform_data(raw_data: list, config: dict) -> dict:
    # Your data processing logic
    filtered = [x for x in raw_data if x > config['threshold']]
    return {
        "filtered_count": len(filtered),
        "data": filtered,
        "metadata": {"threshold": config['threshold']}
    }

# Every call is automatically captured
result = transform_data([1, 5, 10, 2, 8], {"threshold": 4})

Advanced Snapshot Options

@snapshot(
    task_id="advanced_task",
    capture_env=True,        # Capture environment variables
    capture_system=True,     # Capture system info
    storage_backend=None     # Use custom storage
)
def advanced_processing(data):
    return process_complex_data(data)

2. Command Line Interface

Overview & Monitoring

# Comprehensive dashboard overview
dataghost overview

# List all captured snapshots
dataghost snapshot --list

# List snapshots for specific task
dataghost snapshot --task-id my_task

# Show all replayable tasks
dataghost tasks

Debugging & Replay

# Replay latest run of a task
dataghost replay my_task

# Replay specific run
dataghost replay my_task --run-id 20241201_143022

# Replay with validation
dataghost replay my_task --validate

Comparison & Analysis

# Compare latest two runs
dataghost diff my_task

# Compare specific runs
dataghost diff my_task --run-id1 run1 --run-id2 run2

# Output comparison as JSON
dataghost diff my_task --format json

3. Interactive Dashboard

Local Development

# Start dashboard (opens browser automatically)
dataghost dashboard

# Custom port and host
dataghost dashboard --port 3000 --host 0.0.0.0

# Run without opening browser
dataghost dashboard --no-browser

Cloud Environments

# Auto-detects Colab/Jupyter and creates public tunnel
dataghost dashboard --tunnel

# Use specific tunnel service
dataghost dashboard --tunnel --tunnel-service localtunnel

Dashboard Features

• 📊 Real-time Overview: Live statistics and health metrics
• 🎯 Task Health Monitoring: Success rates and performance trends
• ⚡ Recent Activity: Latest pipeline executions with filtering
• 📋 Interactive Task Management: Browse, replay, and compare runs
• 🔄 One-click Operations: Replay tasks directly from the UI
• 📊 Visual Diffs: Side-by-side run comparisons
• 🔍 Detailed Snapshots: Drill down into execution details
• 📈 Performance Analytics: Execution time trends and statistics
• 📱 Mobile Responsive: Works on all devices

4. Programmatic API

Replay Engine

from ttd import ReplayEngine

engine = ReplayEngine()

# Replay latest run
result = engine.replay(task_id="my_task")

# Replay specific run
result = engine.replay(task_id="my_task", run_id="20241201_143022")

# Replay with custom validation
result = engine.replay(task_id="my_task", validate_output=True)

print(f"Replay successful: {result['success']}")
print(f"Original output: {result['original_output']}")
print(f"Replayed output: {result['replayed_output']}")

Diff Engine

from ttd import DiffEngine

diff_engine = DiffEngine()

# Compare latest two runs
diff = diff_engine.diff_task_runs("my_task")

# Compare specific runs
diff = diff_engine.diff_task_runs("my_task", run_id1="run1", run_id2="run2")

# Generate human-readable report
report = diff_engine.generate_diff_report(diff, format="text")
print(report)

Custom Storage

from ttd.storage import DuckDBStorageBackend

# Custom database location
storage = DuckDBStorageBackend(
    db_path="my_pipeline_snapshots.db",
    data_dir="./snapshot_data"
)

# Use with snapshots
@snapshot(task_id="custom_storage", storage_backend=storage)
def my_task(data):
    return process_data(data)

---

🌐 Google Colab & Jupyter

Quick Setup

# Install in Colab
!pip install dataghost[colab]

# Your DataGhost code
from ttd import snapshot

@snapshot(task_id="colab_analysis")
def analyze_data(dataset):
    # Your analysis logic
    return {"mean": sum(dataset) / len(dataset)}

# Run analysis
result = analyze_data([1, 2, 3, 4, 5])

Launch Dashboard

# Auto-detects environment and creates public tunnel
!dataghost dashboard --tunnel

What happens:

1. 🔍 Detects Google Colab environment
2. 🌐 Creates secure ngrok tunnel
3. 📱 Generates public URL (e.g., https://abc123.ngrok.io)
4. 🔗 Share URL with teammates for collaborative debugging

Advanced Colab Usage

# Programmatic setup
from ttd.dashboard.colab_utils import setup_colab_dashboard

public_url, success = setup_colab_dashboard(port=8080)

if success:
    print(f"🚀 Dashboard: {public_url}")
    print("💡 Share this URL with your team!")

---

🏗️ Framework Integrations

Apache Airflow

from ttd.integrations.airflow import DataGhostPythonOperator, create_datahost_dag
from datetime import datetime

# Create DataGhost-enabled DAG
dag = create_datahost_dag(
    dag_id='my_etl_pipeline',
    default_args={'owner': 'data-team'},
    schedule_interval='@daily'
)

# Use DataGhost operators
extract_task = DataGhostPythonOperator(
    task_id='extract_data',
    python_callable=extract_data_function,
    dag=dag
)

transform_task = DataGhostPythonOperator(
    task_id='transform_data', 
    python_callable=transform_data_function,
    dag=dag
)

# Set dependencies
extract_task >> transform_task

Coming Soon

• 🔥 Prefect Integration
• 🚀 Dagster Integration
• 📓 Native Jupyter Support
• 🔧 VS Code Extension

---

🎯 Use Cases

🔍 Debug Pipeline Failures

# When a pipeline fails, replay the exact conditions
from ttd import ReplayEngine

engine = ReplayEngine()
result = engine.replay(task_id="failed_task", run_id="failure_run_id")

if not result['success']:
    print(f"Error: {result['error']}")
    print(f"Inputs: {result['inputs']}")
    print(f"Stack trace: {result['stack_trace']}")

📈 Monitor Performance Changes

# Compare performance between deployments
dataghost diff my_etl_task --run-id1 yesterday --run-id2 today

# See execution time changes, output differences, etc.

🧪 Test Changes Safely

# Test new logic against historical data
historical_inputs = get_historical_inputs("my_task", "specific_run")
new_result = new_function(historical_inputs)

# Compare with historical output
diff_engine = DiffEngine()
diff = diff_engine.compare_outputs(historical_output, new_result)

📊 Data Quality Monitoring

@snapshot(task_id="data_quality_check")
def check_data_quality(df):
    return {
        "row_count": len(df),
        "null_count": df.isnull().sum().sum(),
        "duplicate_count": df.duplicated().sum(),
        "completeness": 1 - (df.isnull().sum().sum() / df.size)
    }

# Track quality metrics over time
quality_result = check_data_quality(daily_data)

---

🔧 Configuration

Environment Variables

export DATAGHOST_DB_PATH="./my_snapshots.db"
export DATAGHOST_DATA_DIR="./my_data"
export DATAGHOST_CAPTURE_ENV="true"
export DATAGHOST_CAPTURE_SYSTEM="true"

Global Settings

from ttd import set_storage_backend
from ttd.storage import DuckDBStorageBackend

# Set global storage backend
set_storage_backend(DuckDBStorageBackend("global.db"))

---

🗄️ Storage Backends

DuckDB (Default)

from ttd.storage import DuckDBStorageBackend

# Default configuration
storage = DuckDBStorageBackend()

# Custom configuration
storage = DuckDBStorageBackend(
    db_path="custom_snapshots.db",
    data_dir="./custom_data"
)

S3 (Coming Soon)

from ttd.storage import S3StorageBackend

storage = S3StorageBackend(
    bucket="my-dataghost-bucket",
    prefix="snapshots/",
    region="us-west-2"
)

---

🚀 Advanced Features

Custom Snapshot Metadata

@snapshot(task_id="custom_meta")
def process_with_metadata(data):
    # Add custom metadata to snapshots
    snapshot_metadata = {
        "data_source": "production_db",
        "processing_mode": "batch",
        "quality_score": calculate_quality(data)
    }
    
    return {
        "result": process_data(data),
        "_metadata": snapshot_metadata
    }

Conditional Snapshots

@snapshot(task_id="conditional", condition=lambda inputs: inputs[0] > 100)
def process_large_datasets(data):
    # Only capture snapshots for large datasets
    return expensive_processing(data)

Performance Optimization

@snapshot(
    task_id="optimized",
    compress_data=True,        # Compress large outputs
    sample_large_data=True,    # Sample large inputs
    max_snapshot_size="10MB"   # Limit snapshot size
)
def memory_efficient_task(large_data):
    return process_efficiently(large_data)

---

🛠️ Development & Contributing

Development Setup

# Clone repository
git clone https://github.com/dataghost/dataghost.git
cd dataghost

# Install development dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run linting
black .
isort .
flake8

Running Examples

# Basic example
python examples/basic_example.py

# Airflow DAG example
python examples/airflow_dag.py

# Google Colab example
python examples/colab_example.py

Contributing Guidelines

1. Fork the repository
2. Create a feature branch
3. Write tests for new functionality
4. Ensure all tests pass
5. Submit a pull request

---

🗺️ Roadmap

✅ v0.1.0 - Core Engine (Current)

• Snapshot decorator with metadata capture
• DuckDB storage backend
• CLI with rich commands
• Replay engine
• Diff engine
• Interactive web dashboard
• Google Colab support

🚧 v0.2.0 - Enhanced Features (In Progress)

• S3 storage backend
• Advanced diff algorithms
• Performance optimizations
• Extended Airflow integration
• Prefect integration

📋 v0.3.0 - Ecosystem Integration

• Dagster integration
• Native Jupyter support
• VS Code extension
• Slack/Teams notifications

🎨 v0.4.0 - Advanced UI

• Advanced dashboard features
• Custom dashboard themes
• Real-time collaboration
• Mobile app

---

📊 Performance

DataGhost is designed for minimal overhead:

• Snapshot capture: < 1ms overhead per function call
• Storage: Efficient compression and deduplication
• Memory usage: < 50MB for typical workloads
• Dashboard: Sub-second response times

---

🤝 Support & Community

Getting Help

• 📚 Documentation: GitHub Wiki
• 💬 Discussions: GitHub Discussions
• 🐛 Issues: GitHub Issues
• 📧 Email: [e-mail]

Community

• ⭐ Star us on GitHub: dataghost/dataghost
• 🐦 Follow updates: @dataghost (coming soon)
• 📺 YouTube tutorials: DataGhost Channel (coming soon)

---

📄 License

MIT License - see LICENSE file for details.

---

🙏 Acknowledgments

• Built with ❤️ by Krish Shah
• Inspired by time-travel debugging concepts from software engineering
• Thanks to the Apache Airflow community for pipeline orchestration patterns
• Special thanks to the Python data engineering community

---

Happy Time-Travel Debugging! 👻✨

Get Started • Documentation • Examples • Contributing