ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Реализовать compaction в векторной БД

1. Цель задачи

Разработать и внедрить механизм compaction (слияния маленьких сегментов) для векторной базы данных на основе Qdrant. Compaction необходим для поддержания стабильной производительности поиска (query latency) при постоянной вставке новых векторов, предотвращая фрагментацию данных и рост количества мелких сегментов. Ключевой результат Query latency (p99) не растет со временем при непрерывной нагрузке на запись и чтение в течение 24 часов.

2. Исходные данные

Перед началом необходимо иметь:

Если нет реального инструмента — симулируем:

1. Создать Python-скрипт, который в цикле вставляет новые векторы (по 1000 за раз) в коллекцию Qdrant с отключенным автоматическим compaction (optimizers_config={default_segment_number: 2, memmap_threshold_kb: 1000}).
2. Параллельно запустить второй скрипт, который выполняет поиск (query) каждые 5 секунд и замеряет latency.
3. Через 30 минут вставок проверить количество сегментов в коллекции через Qdrant API (/collections/{name}) — должно быть > 10.

3. Технологический стек

4. Этапы выполнения

Этап 1: Подготовка окружения и генерация фрагментации (30 минут)

Действия

1. Запустить Qdrant в Docker

   docker run -d --name qdrant -p 6333:6333 qdrant/qdrant:latest
   

2. Создать коллекцию с отключенным compaction

   from qdrant_client import QdrantClient
   from qdrant_client.http.models import VectorParams, OptimizersConfigDiff
   
   client = QdrantClient("localhost", port=6333)
   client.recreate_collection(
       collection_name="test_compaction",
       vectors_config=VectorParams(size=128, distance="Cosine"),
       optimizers_config=OptimizersConfigDiff(
           default_segment_number=2,  # Минимальное количество сегментов
           memmap_threshold_kb=1000,  # Порог для memmap
       )
   )
   

3. Сгенерировать фрагментацию

   import numpy as np
   import time
   
   for i in range(100):  # 100 пакетов по 1000 векторов
       vectors = np.random.rand(1000, 128).tolist()
       ids = list(range(i*1000, (i+1)*1000))
       client.upsert(
           collection_name="test_compaction",
           points=[{"id": idx, "vector": vec} for idx, vec in zip(ids, vectors)]
       )
       time.sleep(.src)  # Имитация реальной нагрузки
   

4. Проверить количество сегментов

   collection_info = client.get_collection("test_compaction")
   print(f"Количество сегментов: {collection_info.segments_count}")
   

Ожидаемый результат этапа Коллекция с 10+ сегментами и измеренным baseline latency.

Этап 2: Разработка механизма compaction (2 часа)

Действия

1. Написать функцию для определения кандидатов на compaction:

   def find_compaction_candidates(client, collection_name, max_segments=5):
       collection_info = client.get_collection(collection_name)
       segments = collection_info.segments
       
       # Критерии: маленькие сегменты (< 1000 векторов)
       candidates = []
       for segment in segments:
           if segment.num_vectors < 1000:
               candidates.append(segment.id)
       
       return candidates if len(candidates) > 1 else []
   

2. Реализовать функцию compaction

   def compact_segments(client, collection_name, segment_ids):
       # 1. Извлечь все векторы из указанных сегментов
       all_points = []
       for seg_id in segment_ids:
           points = client.scroll(
               collection_name=collection_name,
               segment_id=seg_id,
               limit=10000
           )
           all_points.extend(points[0])
       
       # 2. Удалить старые сегменты (через API Qdrant)
       for seg_id in segment_ids:
           client.delete_segment(collection_name, seg_id)
       
       # 3. Вставить все векторы в новый сегмент
       client.upsert(
           collection_name=collection_name,
           points=[{"id": p.id, "vector": p.vector} for p in all_points]
       )
   

3. Добавить оптимизации

   • Использовать batch-операции для извлечения/вставки
   • Реализовать блокировку поиска во время compaction (read-only mode)
   • Добавить проверку целостности данных после compaction

Ожидаемый результат этапа Рабочая функция compaction, которая объединяет маленькие сегменты в один.

Этап 3: Интеграция compaction в цикл обработки (1 час)

Действия

1. Создать фоновый поток для compaction

   import threading
   import time
   
   def compaction_worker(client, collection_name, interval=300):
       while True:
           time.sleep(interval)
           candidates = find_compaction_candidates(client, collection_name)
           if candidates:
               print(f"Compacting segments: {candidates}")
               compact_segments(client, collection_name, candidates)
               print("Compaction completed")
   

2. Запустить compaction в отдельном потоке

   compaction_thread = threading.Thread(
       target=compaction_worker,
       args=(client, "test_compaction"),
       daemon=True
   )
   compaction_thread.start()
   

3. Модифицировать скрипт нагрузки

   • Добавить замер latency до и после compaction
   • Логировать количество сегментов каждые 10 минут

Ожидаемый результат этапа Compaction запускается автоматически каждые 5 минут и объединяет маленькие сегменты.

Этап 4: Тестирование и измерение производительности (1.5 часа)

Действия

1. Запустить нагрузочное тестирование

   import time
   import numpy as np
   
   latencies = []
   start_time = time.time()
   
   while time.time() - start_time < 3600:  # 1 час
       query_vector = np.random.rand(128).tolist()
       t0 = time.time()
       client.search(
           collection_name="test_compaction",
           query_vector=query_vector,
           limit=10
       )
       latency = time.time() - t0
       latencies.append(latency)
       
       # Вставка новых векторов
       if len(latencies) % 10 == 0:
           vectors = np.random.rand(100, 128).tolist()
           ids = list(range(len(latencies)*100, (len(latencies)+1)*100))
           client.upsert(
               collection_name="test_compaction",
               points=[{"id": idx, "vector": vec} for idx, vec in zip(ids, vectors)]
           )
       
       time.sleep(0.5)
   

2. Визуализировать результаты

   import matplotlib.pyplot as plt
   import pandas as pd
   
   df = pd.DataFrame({"latency": latencies})
   df["time"] = pd.date_range(start="2025-01-01", periods=len(latencies), freq="S")
   
   plt.figure(figsize=(12, 6))
   plt.plot(df["time"], df["latency"], alpha=0.5, label="Raw latency")
   plt.axhline(y=np.percentile(latencies, 99), color='r', linestyle='--', label="p99")
   plt.xlabel("Time")
   plt.ylabel("Latency (seconds)")
   plt.title("Query Latency Over Time with Compaction")
   plt.legend()
   plt.savefig("latency_with_compaction.png")
   

3. Сравнить с baseline (без compaction):

   • Запустить тест без compaction на 30 минут
   • Сравнить p99 latency в конце теста

Ожидаемый результат этапа График latency, показывающий стабильность p99 на уровне < 50ms.

Этап 5: Оптимизация и настройка параметров (1 час)

Действия

1. Настроить параметры compaction

   Параметр	Значение по умолчанию	Оптимальное значение	Обоснование
   max_segments	5	3	Меньше сегментов — быстрее поиск
   min_vectors_for_compaction	1000	500	Более агрессивное слияние
   compaction_interval	300s	120s	Чаще проверять фрагментацию

2. Реализовать адаптивный compaction

   def adaptive_compaction(client, collection_name):
       collection_info = client.get_collection(collection_name)
       segments = collection_info.segments
       
       # Если сегментов больше порога — запускаем compaction
       if len(segments) > 5:
           # Сортируем сегменты по размеру
           sorted_segments = sorted(segments, key=lambda s: s.num_vectors)
           # Объединяем самые маленькие
           candidates = [s.id for s in sorted_segments[:3]]
           compact_segments(client, collection_name, candidates)
   

3. Добавить мониторинг в реальном времени

   • Экспортировать метрики в Prometheus (опционально)
   • Логировать события compaction в файл

Ожидаемый результат этапа Настроенный механизм compaction с адаптивными параметрами.

5. Критерии приемки (Definition of Done)

• Qdrant запущен и коллекция создана с отключенным автоматическим compaction
• Сгенерирована фрагментация (10+ сегментов)
• Реализована функция find_compaction_candidates, корректно определяющая маленькие сегменты
• Реализована функция compact_segments, объединяющая сегменты без потери данных
• Compaction запускается автоматически в фоновом потоке
• Query latency (p99) не превышает 50ms в течение 1 часа теста
• Количество сегментов не превышает 5 после compaction
• Данные не теряются при compaction (проверка через поиск известных векторов)
• Написан скрипт для визуализации latency
• Сравнительный график "с compaction" vs "без compaction" показывает улучшение

6. Ожидаемый результат

Основной артефакт Python-скрипт compaction_demo.py, который:

• Запускает Qdrant (если не запущен)
• Создаёт коллекцию с отключенным автоматическим compaction
• Генерирует фрагментацию
• Запускает фоновый compaction
• Выполняет нагрузочное тестирование в течение 1 часа
• Сохраняет график latency в latency_with_compaction.png

Дополнительные результаты

• Лог-файл compaction.log с событиями compaction
• Сравнительный график latency_comparison.png
• Отчёт report.md с анализом производительности

7. Возможные сложности и их решение

8. Бюджет времени (оценка)

Примечание Для первого раза рекомендуется заложить +2 часа на отладку и изучение документации Qdrant.

9. Связанные вопросы из базы знаний

10. Чек-лист самопроверки

• Я проверил, что Qdrant запущен и доступен по порту 6333
• Я убедился, что коллекция создана с правильными параметрами (default_segment_number=2)
• Я сгенерировал минимум 10000 векторов, чтобы создать фрагментацию
• Я проверил, что количество сегментов > 5 перед запуском compaction
• Я запустил compaction и убедился, что количество сегментов уменьшилось
• Я замерил latency до и после compaction и сравнил результаты
• Я сохранил график latency и лог compaction
• Я проверил, что данные не потерялись (поиск по ID)
• Я написал отчёт с анализом производительности