Некоторое время назад наш поиск стал работать быстрее. Особенно это заметно на сложных для движка запросах, в которых используется минимум фильтров и высокочастотные слова, что требует построить фасеты по результатам и отсортировать максимальные объёмы документов. Но и запросы средней сложности, где в выдаче немного документов, стали обрабатываться заметно быстрее. Почему возникла необходимость что-то ускорять и как мы это делали?
Поиск на hh.ru – это:
- 400 запросов в секунду;
- 26 гигабайт обновляющегося в realtime индекса;
- 3-кратный коэффициент репликации (резервирование отказоустойчивости);
- 5-кратный запас по производительности.
И вся эта прелесть при общей загрузке системы в 15% на некоторых запросах работала непростительно медленно. Ввиду того, что «активный» индекс резюме значительно больше остальных, особенно это было критично для работодателей.
В основе поиска hh.ru лежит Lucene, поверх которой у нас за много лет написано достаточно много кода. Ранее мы решали частные задачи по оптимизации, в рамках которых пришли к пониманию, что производительность поиска упирается в производительность Lucene. Точнее в то, как мы её используем.
Известно, что то, что нельзя ускорить «в лоб», часто можно распараллелить. В Lucene с версии 3.1.0 имелась возможность делить каждый запрос на несколько потоков по числу сегментов. Но рядом имелся (и в 4.3 версии имеется) комментарий «TODO: should we make this threaded…? the Collector could be sync’d? always use single thread».
А коллекторы (механизм, получающий «по одному» все найденные документы в сегменте) у нас используются повсеместно: на них основан наш код фасетов и сортировок.
В факультативном порядке мы провели эксперимент, в рамках которого был изолирован код, завязанный на коллекторах, данные разбиты на некоторое число сегментов, и произведено сравнение с линейным и параллельным поиском. Он подтвердил возможность ускорения, поэтому мы спланировали задачу, и работа закипела.
Общий план выглядел так:
- добавляем коллекторам возможность сбора и объединения результатов из нескольких сегментов;
- реализуем/включаем параллельный обход сегментов;
- разбиваем индекс на N равных по размеру сегментов;
- PROFIT.
Пункт 1. Коллекторы
Получилось, что первый пункт плана был реализован в результате рефакторинга в задаче, не имевшей прямого отношения к распараллеливанию. В результате мы получили механизм, позволяющий объединять дерево результатов поиска на скольких угодно уровнях (сейчас у нас четыре уровня: индексы по типам документов, индексов, реплики, сегменты).
Пункт 2. Параллельный обход сегментов
На нём остановлюсь подробнее.
Нам крайне важен был следующий метод IndexSearcher’а:
public void search(Weight weight, Filter filter, Collector collector)
throws IOException {
// TODO: should we make this
// threaded...? the Collector could be sync'd?
// always use single thread:
for (int i = 0; i < subReaders.length; i++) { // search each subreader
collector.setNextReader(subReaders[i], docBase + docStarts[i]);
final Scorer scorer = (filter == null) ?
weight.scorer(subReaders[i], !collector.acceptsDocsOutOfOrder(), true) :
FilteredQuery.getFilteredScorer(subReaders[i], getSimilarity(), weight, weight, filter);
if (scorer != null) {
scorer.score(collector);
}
}
}
Здесь в цикле по сегментам коллектор переключается на очередной из списка collector.setNextReader(…), а затем scorer «скармливает» найденные документы в коллектор. Именно переключение на сегмент и лишает нас всей прелести многопоточности: при параллельном поиске коллектор не будет знать, к какому сегменту относится тот или иной документ. Решение оказалось достаточно простым: сделать суперколлектор, который будет создавать работников на каждый сегмент:
public interface ParallelCollector {
/**
* creates per-segment collector
*/
Collector createPartialCollector();
}
С таким подходом модификация IndexSearcher’а вышла простой:
- передаём наш «коллектор»;
- цикл по subReaders;
- получаем и инициализируем выделенный коллектор:
Collector collector = parallelCollector.createPartialCollector(); collector.setNextReader(subReader, subreaderDocBase); - уже в Runnable выполняем имеющийся старый код и submit’им его в имеющийся executor;
- отлавливаем ошибки. В нашем случае – отлавливаем и возвращаем наружу первое полученное исключение.
public void search(final Weight weight, final Filter filter, final ParallelCollector parallelCollector) throws IOException {
final CountDownLatch latch = new CountDownLatch(subReaders.length);
final AtomicReference<Throwable> exceptionReference = new AtomicReference<Throwable>();
for (int i = 0; i < subReaders.length; i++) {
final int subreaderDocBase = docStarts[i];
final IndexReader subReader = subReaders[i];
executor.submit(new Runnable() {
@Override
public void run() {
try {
Collector collector = parallelCollector.createPartialCollector();
collector.setNextReader(subReader, subreaderDocBase);
Scorer scorer = (filter == null) ?
weight.scorer(subReader, !collector.acceptsDocsOutOfOrder(), true) :
FilteredQuery.getFilteredScorer(subReader, getSimilarity(), weight, weight, filter);
if (scorer != null) {
scorer.score(collector);
}
} catch (Throwable t) {
exceptionReference.compareAndSet(null, t);
throw new RuntimeException(t);
} finally {
latch.countDown();
}
}
});
}
try {
latch.await();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
Throwable possibleException = exceptionReference.get();
if (possibleException != null) {
if (possibleException instanceof RuntimeException) {
throw (RuntimeException) possibleException;
} else
if (possibleException instanceof IOException) {
throw (IOException) possibleException;
} else {
throw new RuntimeException(possibleException);
}
}
}
Пункт 3. Разбивка сегментов
В Lucene по умолчанию предполагается, что сегмент должен быть один. Точнее, к этому Люсина стремится. На самом деле, на каждый flush данных на диск создаётся новый маленький сегмент, который дальше, в соответствии с MergePolicy, автоматически объединяется с другими маленькими сегментами в более крупные, и так по нарастающей. При работающем обновлении индекса «хвост» из мелких сегментов присутствует всегда.
Но разработчики молодцы: они дали средство для ограничения максимального размера сегмента, чем мы и воспользовались — setMaxMergeMB + setMaxMergeMBForForcedMerge решило задачу на ура.
Бонусом решения 3-го пункта стало избавление от механизма оптимизации индекса. В Lucene документы в индекс дописываются. Если требуется документ переиндексировать, старый помечается удалённым, а новая его версия дописывается в конец индекса. В результате со временем появляется много «дыр», индекс раздувается, из-за чего сильно снижается производительность.
Бороться с этим можно периодическими mergeDeletes (ранее expungeDeletes) и forceMerge (ранее optimize), которые копируют «живые» документы из старого (возможно, нескольких) в новый сегмент. Операции эта довольно дорогие в плане дискового ввода/вывода и расхода памяти.
Сегменты малого размера заметно снижают затраты ввиду того, что для обновления/удаления документов приходится переписывать меньше соседних документов.
Результат
Итак, за почти месяц разработки мы получили параллельный поиск и много небольших сегментов. В чем ценность этого?
- Более быстрый поиск. Теперь результат 95 % поисковых запросов по вакансиям выдается за 10 миллисекунд, а по резюме – за 70 миллисекунд. Для сравнения, еще несколько месяцев назад это было 30 и 270 соответственно.
- Возможность предложить патч в Lucene (уже вот-вот, но хочется «причесать код»).
- Избавление от дополнительного механизма оптимизации индекса.
Наглядный результат
Интервал – 2 недели.
Красная линия – было, синяя – стало, ось Y – время отклика.
50-я квантиль, поисковые запросы средней сложности:
И 95-я квантиль, сложные для поиска запросы с максимальным числом результатов:
