> Как распараллелить обновление данных для миллиона пользователей за ограниченное время? (Python)

Короткий ответ

Основная стратегия - разбить задачу на независимые батчи и обрабатывать их конкурентно через пул воркеров, используя asyncio для I/O-bound операций или multiprocessing для CPU-bound. Ключевые элементы: чанкование по ID пользователей, ограничение concurrency через semaphore, ретраи с exponential backoff и идемпотентность операций. Для миллиона пользователей важно не упираться в rate limits внешних сервисов и не перегружать БД - поэтому батчи должны быть небольшими (100-500), а параллелизм - настраиваемым.

Подробное объяснение

Задача сводится к классическому паттерну producer-consumer с ограниченным пулом воркеров. Миллион пользователей - это не так много для современных систем, если операция I/O-bound (HTTP-запросы, запись в БД). Проблема не в количестве, а в ограничениях: rate limits API, лимиты соединений к БД, память.

Основные подходы:

1. Чанкование: разбиваем пользователей на батчи по 100-500 записей. Это позволяет контролировать нагрузку и прогресс.
2. Concurrency control: используем asyncio.Semaphore или ThreadPoolExecutor с фиксированным числом воркеров (обычно 10-50 для I/O). Для CPU-bound - ProcessPoolExecutor с числом процессов = числу ядер.
3. Идемпотентность: каждая операция должна быть безопасна для повторного выполнения - если воркер упал, мы можем перезапустить батч без побочных эффектов.
4. Обработка ошибок: ретраи с exponential backoff (например, 1s, 2s, 4s) и максимум 3-5 попыток. Неудачные батчи складываем в очередь на повторную обработку.
5. Прогресс и observability: логируем количество обработанных, ошибки, время выполнения. Используем tqdm или кастомный счётчик.

Если время жёстко ограничено (например, 10 минут), нужно рассчитать пропускную способность: 1M / 600s ≈ 1667 операций/сек. При батче 100 это 17 батчей/сек. Один воркер с latency 50ms обрабатывает 20 операций/сек - значит нужно минимум 84 воркера. Это реалистично для asyncio.

На практике

Типичный сценарий: нужно обновить поле last_seen или отправить push-уведомление. Сначала получаем список ID из БД (лучше через streaming cursor, чтобы не держать миллион в памяти). Затем обрабатываем батчами.

Важно: не создавать миллион корутин или потоков - это убьёт память. Всегда через пул с ограничением. Для БД используем connection pool с размером, равным числу воркеров.

Если операция включает внешний API - обязательно проверяем rate limits и делаем адаптивную паузу. Если API отдаёт 429 - увеличиваем задержку.

Для больших объёмов можно использовать очередь (например, asyncio.Queue), но для миллиона это избыточно - достаточно простого цикла по чанкам с gather и ограничением.

Пример кода

PYTHON
import asyncio
import aiohttp
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy import text

BATCH_SIZE = 200
CONCURRENCY = 50
MAX_RETRIES = 3

async def process_user(session: AsyncSession, user_id: int, sem: asyncio.Semaphore):
    async with sem:
        for attempt in range(MAX_RETRIES):
            try:
                # I/O-bound операция: например, вызов внешнего API
                async with aiohttp.ClientSession() as http:
                    async with http.post(f"https://api.example.com/update/{user_id}") as resp:
                        resp.raise_for_status()
                # Обновляем в БД
                await session.execute(
                    text("UPDATE users SET updated_at = NOW() WHERE id = :id"),
                    {"id": user_id}
                )
                await session.commit()
                return
            except Exception:
                if attempt == MAX_RETRIES - 1:
                    raise
                await asyncio.sleep(2 ** attempt)

async def main():
    engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
    async with engine.connect() as conn:
        # Streaming cursor, чтобы не грузить все ID в память
        result = await conn.stream(text("SELECT id FROM users"))
        sem = asyncio.Semaphore(CONCURRENCY)
        tasks = []
        async for row in result:
            tasks.append(process_user(conn, row.id, sem))
            if len(tasks) >= BATCH_SIZE:
                await asyncio.gather(*tasks, return_exceptions=True)
                tasks = []
        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)

asyncio.run(main())

Как отвечать на собеседовании

Начни с уточнения: операция I/O-bound или CPU-bound? Это определяет выбор между asyncio и multiprocessing. Затем опиши архитектуру: чанки → пул воркеров → обработка ошибок. Обязательно упомяни ограничения: rate limits, connection pool, память. Покажи, что понимаешь trade-off между concurrency и нагрузкой на БД.

Если спросят про время - приведи расчёт: сколько операций в секунду нужно и сколько воркеров для этого потребуется. Это демонстрирует инженерное мышление.

Хорошо добавить: "Я бы сначала сделал нагрузочный тест на 10k пользователей, чтобы подобрать оптимальный BATCH_SIZE и CONCURRENCY, а потом масштабировал".

Что проверяет интервьюер

• Понимание разницы между I/O-bound и CPU-bound задачами.
• Умение проектировать конкурентные системы с ограничением ресурсов.
• Знание паттернов обработки ошибок: retry, backoff, идемпотентность.
• Практический опыт с asyncio, пулами, streaming.
• Умение оценивать пропускную способность и масштабируемость.
• Внимание к деталям: не создавать миллион корутин, не блокировать event loop.

Типичные ошибки

• Создание миллиона задач через asyncio.gather без ограничения - память и сокеты закончатся.
• Игнорирование rate limits внешнего API - получите бан или 429.
• Отсутствие ретраев - один сбой в сети убьёт весь батч.
• Загрузка всех ID в память списком - для миллиона это 8-16 MB, но лучше использовать streaming.
• Не учитывать connection pool БД - если воркеров 100, а pool 10, будет contention.
• Синхронные вызовы внутри async-функций - блокируют event loop.
• Отсутствие идемпотентности - повторный запуск приведёт к дублированию операций.