Tools / Libraries / etc for running large migrations (i.e. managing optimal number of concurrent workers) ?

[jvanasco]

I need to run a fairly large database migration (5BN+ rows).

When I've done this in the past, I've done the following:

• Create a first table of the record IDs to migrate : record_id[int], processing_status[bool - todo, locked, migrated]
• Create a second table to track migration performance: serial, worker_id, timestamp, number of records processed, batch duration
• Use a windowed range to pull IDs ( https://github.com/sqlalchemy/sqlalchemy/wiki/RangeQuery-and-WindowedRangeQuery ) for each batch
• Manually tweak the number of migration scripts running at a given time, by querying the performance table and adding/removing workers

Has anyone built a tool for doing things like this, or is there a better pattern? The optimal number of workers is wildly different on the dev, ci, staging and production servers.

[mr-adonis-jimenez]

For managing large-scale migrations (5BN+ rows) with adaptive worker concurrency, here are battle-tested tools and patterns:

Production-Ready Tools

1. Celery + Redis (Most Common Production Pattern)

Best for dynamic worker scaling:

from celery import Celery, group
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

app = Celery('migration', broker='redis://localhost:6379/0')

@app.task(bind=True, max_retries=3)
def migrate_batch(self, start_id, end_id):
    engine = create_engine('postgresql://...')
    with Session(engine) as session:
        try:
            # Your migration logic with windowed range
            records = session.execute(
                text("SELECT * FROM source WHERE id BETWEEN :start AND :end FOR UPDATE SKIP LOCKED"),
                {"start": start_id, "end": end_id}
            )
            # Process batch
            session.commit()
            return {"processed": end_id - start_id, "duration": ...}
        except Exception as exc:
            session.rollback()
            raise self.retry(exc=exc, countdown=60)

# Adaptive scaling based on throughput
def run_migration_adaptive():
    batch_size = 10000
    max_workers = get_optimal_workers()  # Start with DB connection limit
    
    job = group(
        migrate_batch.s(i, i + batch_size) 
        for i in range(0, 5_000_000_000, batch_size)
    )
    result = job.apply_async()
    return result

Dynamic scaling with Flower monitoring:

celery -A migration worker --autoscale=10,3 --loglevel=info
celery -A migration flower  # Web UI at localhost:5555

2. Luigi (Dependency-Based Pipeline)

Good for complex migration dependencies:

import luigi
from luigi.contrib.postgres import CopyToTable

class MigrationBatch(luigi.Task):
    batch_id = luigi.IntParameter()
    
    def run(self):
        # Track performance
        start = time.time()
        with self.output().open('w') as f:
            # Migration logic
            f.write(f"{self.batch_id},{time.time()-start}\n")
    
    def output(self):
        return luigi.LocalTarget(f'/tmp/migration_{self.batch_id}.csv')

class MigrationPipeline(luigi.WrapperTask):
    def requires(self):
        # Automatically manages parallelism
        return [MigrationBatch(i) for i in range(500000)]

3. Airflow (Enterprise-Grade)

For production with monitoring:

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook

def migrate_batch_airflow(start_id, end_id, **context):
    hook = PostgresHook(postgres_conn_id='prod_db')
    # Migration logic with connection pooling
    pass

with DAG('migration_pipeline', 
         max_active_runs=1,
         concurrency=16) as dag:  # Auto-scales
    
    tasks = [
        PythonOperator(
            task_id=f'migrate_{i}',
            python_callable=migrate_batch_airflow,
            op_args=[i*10000, (i+1)*10000],
            pool='migration_pool'  # Define resource pools
        )
        for i in range(500000)
    ]

4. RQ (Redis Queue) - Lightweight Alternative

from rq import Queue
from redis import Redis

redis_conn = Redis()
q = Queue('migrations', connection=redis_conn)

# Enqueue batches
for batch in generate_batches(5_000_000_000, batch_size=50000):
    job = q.enqueue(
        migrate_batch,
        batch['start'], 
        batch['end'],
        job_timeout='10m',
        result_ttl=86400
    )

# Workers auto-scale based on queue depth
# rq worker migrations --burst --worker-class rq.worker.SimpleWorker

Adaptive Worker Management Pattern

import psutil
from sqlalchemy import create_engine, text

class AdaptiveMigrationManager:
    def __init__(self, db_url):
        self.engine = create_engine(db_url, pool_size=20, max_overflow=10)
        self.performance_data = []
    
    def get_optimal_workers(self):
        """Dynamically calculate optimal worker count"""
        # Check database connections
        with self.engine.connect() as conn:
            active_conns = conn.execute(
                text("SELECT count(*) FROM pg_stat_activity WHERE state = 'active'")
            ).scalar()
        
        max_db_conns = 100  # From pg_max_connections
        available_conns = max_db_conns - active_conns - 10  # Reserve 10
        
        # Check system resources
        cpu_percent = psutil.cpu_percent(interval=1)
        available_cpus = psutil.cpu_count()
        memory_percent = psutil.virtual_memory().percent
        
        # Calculate based on bottleneck
        cpu_workers = max(1, int(available_cpus * (1 - cpu_percent/100)))
        memory_workers = 20 if memory_percent < 80 else 10
        db_workers = max(5, available_conns // 2)
        
        return min(cpu_workers, memory_workers, db_workers)
    
    def adjust_workers_dynamically(self, current_workers):
        """Monitor and adjust based on throughput"""
        recent_perf = self.performance_data[-10:]  # Last 10 batches
        avg_duration = sum(p['duration'] for p in recent_perf) / len(recent_perf)
        
        if avg_duration > 60:  # Batches taking too long
            return max(1, current_workers - 2)
        elif avg_duration < 10:  # Can handle more
            return min(50, current_workers + 5)
        return current_workers

SQLAlchemy-Specific Optimization

from sqlalchemy import event
from sqlalchemy.pool import NullPool

# Use NullPool for migration workers to avoid connection pool exhaustion
engine = create_engine(
    'postgresql://...',
    poolclass=NullPool,  # Each worker manages own connection
    execution_options={
        "isolation_level": "READ COMMITTED"
    }
)

# Skip locked rows for concurrent processing
from sqlalchemy.dialects.postgresql import insert

def get_next_batch(session, batch_size):
    """SKIP LOCKED prevents worker contention"""
    return session.execute(
        text("""
            UPDATE migration_queue 
            SET status = 'locked', worker_id = :worker
            WHERE id IN (
                SELECT id FROM migration_queue 
                WHERE status = 'pending'
                ORDER BY id
                LIMIT :limit
                FOR UPDATE SKIP LOCKED
            )
            RETURNING id
        """),
        {"worker": worker_id, "limit": batch_size}
    ).fetchall()

Recommended Architecture

For your 5BN row use case:

1. Queue System: Celery with Redis
2. Monitoring: Prometheus + Grafana for metrics
3. Worker Scaling: Start with db_connection_limit / 3 workers
4. Batch Size: 50K-100K rows per batch (adjust based on row size)
5. Coordination: Use FOR UPDATE SKIP LOCKED for lock-free batching

# Complete production example
from celery import Celery
from prometheus_client import Counter, Histogram

app = Celery('migration')

migration_counter = Counter('migration_rows_processed', 'Rows migrated')
migration_duration = Histogram('migration_batch_duration', 'Batch duration')

@app.task
def migrate_with_metrics(start_id, end_id):
    with migration_duration.time():
        rows_processed = perform_migration(start_id, end_id)
        migration_counter.inc(rows_processed)

Key Differences from Your Current Approach

• ✅ Auto-scaling: Tools handle worker management
• ✅ Retry logic: Built-in failure handling
• ✅ Monitoring: Web UIs and metrics out-of-box
• ✅ Distributed: Run across multiple servers
• ✅ Backpressure: Queue systems prevent overload

For dev/staging/prod differences, use environment-based worker configs:

WORKER_CONFIG = {
    'dev': {'workers': 2, 'batch_size': 1000},
    'staging': {'workers': 8, 'batch_size': 50000},
    'prod': {'workers': 32, 'batch_size': 100000}
}

[jvanasco]

The comments and code above look like they were entirely generated by an AI prompt.

"For managing large-scale migrations (5BN+ rows) with adaptive worker concurrency, here are battle-tested tools and patterns:"

Can you please qualify "battle tested" ? Did you personally use these tools and patterns? Did you write any of this code?