Высокая производительность — одно из ключевых требований при работе с большими данными. Мы в управлении загрузки данных в Сбере занимаемся прокачкой практически всех транзакций в наше Облако Данных на базе Hadoop и поэтому имеем дело с действительно большими потоками информации. Естественно, что мы все время ищем способы повысить производительность, и теперь хотим рассказать, как удалось пропатчить RegionServer HBase и HDFS-клиент, благодаря чему удалось значительно увеличить скорость операции чтения.


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


Таким образом далее мы будем разбираться с ускорением доступа только к тем данным, которые есть в кэше ОС или находятся в сравнимых по скорости доступа хранилищ типа SSD/NVMe.

В нашем случае мы будем проводить тесты на стенде из 4х серверов, каждый из которых заряжен следующим образом:

CPU: Xeon E5-2680 v4 @ 2.40GHz 64 threads.
Память: 730 Гб.
java version: 1.8.0_111

И тут собственно ключевой момент — объем данных в таблицах, которые требуется вычитывать. Дело в том, что если читать данные из таблицы, которая целиком помещаются в кэш HBase, то до чтения из buff/cache операционки дело даже не дойдет. Потому что HBase по умолчанию выделяет 40% памяти под структуру которая называется BlockCache. По сути это ConcurrentHashMap, где ключ это имя файла+offset блока, а value собственно данные по этому смещению.

Таким образом, когда чтение идет только из этой структуры, мы видим великолепную скорость, вроде миллиона запросов в секунду. Но давайте представим себе, что мы не можем отдавать сотни гигабайт памяти только под нужды БД, потому что на этих серверах крутится много чего еще полезного.

Например в нашем случае объем BlockCache на одном RS это порядка 12 Гб. Мы высадили два RS на одну ноду, т.е. под BlockCache выделено 96 Гб на всех нодах. А данных при этом во много раз больше, например пусть это будет 4 таблицы, по 130 регионов, в которых файлы размером по 800 Мб, пожатые FAST_DIFF, т.е. в сумме 410 Гб (это чистые данные, т.е. без учета фактора репликации).

Таким образом, BlockCache составляет лишь около 23% от общего объема данных и это гораздо ближе к реальным условиям того, что называется BigData. И вот тут начинается самое интересное — ведь очевидно, чем меньше попаданий в кэш, тем хуже производительность. Ведь в случае промаха придется выполнить кучу работы — т.е. спуститься до вызова системных функций. Однако этого не избежать и поэтому давайте рассмотрим совсем другой аспект — а что происходит с данными внутри кэша?

Упростим ситуацию и допустим, что у нас есть кэш в который помещается только 1 объект. Вот пример того что произойдет при попытке работы с объемом данных в 3 раза больше чем кэш, нам придется:

1. Поместить блок 1 в кэш
2. Удалить блок 1 из кэша
3. Поместить блок 2 в кэш
4. Удалить блок 2 из кэша
5. Поместить блок 3 в кэш

Проделано 5 действий! Однако нормальной этой ситуацию называть никак нельзя, по сути мы заставляем HBase проделывать кучу совершенно бесполезной работы. Он постоянно вычитывает данные из кэша ОС, помещает его себе в BlockCache, для того чтобы почти тут же выкинуть его, потому что приехала новая порция данных. Анимация в начале поста показывает суть проблемы — Garbage Collector зашкаливает, атмосфера греется, маленькая Грета в далекой и жаркой Швеции расстраивается. А мы айтишники очень не любим, когда грустят дети, поэтому начинаем думать, что с этим можно поделать.

А что если помещать в кэш не все блоки, а только определенный процент из них, так чтобы кэш не переполнялся? Давайте для начала просто добавим всего несколько строк кода в начало функции помещения данных в BlockCache:

  public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory) {
    if (cacheDataBlockPercent != 100 && buf.getBlockType().isData()) {
      if (cacheKey.getOffset() % 100 >= cacheDataBlockPercent) {
        return;
      }
    }
...

Смысл тут в следующем, оффсет — это положение блока в файле и последние цифры его случайно и равномерно распределены от 00 до 99. Поэтому мы будем пропускать только те, которые попадают в нужный нам диапазон.

Например выставим cacheDataBlockPercent = 20 и посмотрим что будет:



Результат налицо. На графиках ниже становится понятно, за счет чего произошло такое ускорение — мы экономим кучу ресурсов GC не занимаясь сизифовым трудом размещения данных к кэше лишь для того, чтобы тут же выбросить их в марсианским псам под хвост:



Утилизация CPU при этом растет, однако сильно меньше чем производительность:



Тут еще стоит отметить, что блоки которые хранятся в BlockCache бывают разные. Большая часть, порядка 95% это собственно данные. А остальное это метаданных, типа Bloom фильтров или LEAF_INDEX и т.д.. Этих данных мало, но они очень полезные, так как прежде чем обратиться непосредственно к данным, HBase обращается к мете, чтобы понять нужно ли искать тут дальше и если да, то где именно находится интересующий его блок.

Поэтому в коде мы видим условие проверку buf.getBlockType().isData() и благодаря этому мету мы будем оставлять в кэше в любом случае.

Теперь давайте увеличим нагрузку и за одно слегка затюним фичу. В первом тесте мы сделали процент отсечения = 20 и BlockCache был немного недозагружен. Теперь поставим 23% и будем добавлять по 100 потоков каждые 5 минут, чтобы увидеть, в какой момент происходит насыщение:



Тут мы видим, что исходная версия практически сразу упирается в потолок на уровне около 100 тыс запросов в секунду. Тогда как патч дает ускорение до 300 тысяч. При этом понятно, что дальнейшее ускорение уже не такое «бесплатное», утилизация CPU при этом тоже растет.

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

Для управления этим было добавлено три параметра:

hbase.lru.cache.heavy.eviction.count.limit — устанавливает, сколько раз должен запуститься процесс выселения данных из кеша, прежде чем мы начнем использовать оптимизацию (т.е. пропускать блоки). По умолчанию оно равно MAX_INT = 2147483647 и фактически означает, что фича никогда не начнет работать при таком значении. Потому что процесс выселения запускается каждые 5 — 10 секунд (это зависит от нагрузки) и 2147483647 * 10 / 60 / 60 / 24 / 365 = 680 лет. Однако мы можем установить этот параметр равным 0 и заставить фичу работать сразу же после старта.

Однако есть и полезная нагрузка в этом параметре. Если у нас характер нагрузки такой, что постоянно перемежаются краткосрочные чтения (допустим днем) и долгосрочные (по ночам), то мы можем сделать так, что фича будет включаться только когда идут продолжительные операции чтения.

Например мы знаем, что краткосрочные чтения длятся обычно около 1 минуты. Нам не надо начинать выкидывать блоки, кеш не успеет устареть и тогда мы можем установить этот параметр равным например 10. Это приведет к тому, что оптимизация начнет работать только когда началось длительное активное чтение, т.е. через 100 секунд. Таким образом если мы имеем краткосрочное чтение, то все блоки попадут в кеш и будут доступны (за исключением тех что будут выселены стандартным алгоритмом). А когда мы делаем долгосрочные чтения, фича включается и бы имеем намного более высокую производительность.

hbase.lru.cache.heavy.eviction.mb.size.limit — устанавливает, как много мегабайт нам хотелось бы помещать в кеш (и естественно выселять) за 10 секунд. Фича будет пытаться достигнуть этого значения и поддерживать его. Смысл в следующем, если мы пихаем в кеш гигабайты, то и выселять придется гигабайты, а это, как мы видели выше, весьма накладно. Однако не нужно пытаться выставить его слишком маленьким, так как это приведет к преждевременному выходу из режима пропуска блоков. Для мощных серверов (порядка 20-40 физических ядер) оптимально выставлять около 300-400 МБ. Для среднего класса (~10 ядер) 200-300 МБ. Для слабых систем (2-5 ядра) может быть нормально 50-100 МБ (на таких не тестировалось).

Рассмотрим, как это работает: допустим мы выставили hbase.lru.cache.heavy.eviction.mb.size.limit = 500, идет какая-то нагрузка (чтения) и тогда каждые ~10 секунд мы вычисляем, сколько байт было выселено по формуле:

Overhead = Freed Bytes Sum (MB) * 100 / Limit (MB) — 100;

Если по факту было выселено 2000 MB, то Overhead получается равным:

2000 * 100 / 500 — 100 = 300%

Алгоритмы же стараются поддерживать не больше чем несколько десятков процентов, так что фича будет уменьшать процент кешируемых блоков, тем самым реализуя механизм авто-тюнинга.

Однако если нагрузка упала, допустим выселено всего 200 МБ и Overhead стал отрицательным (так называемый overshooting):

200 * 100 / 500 — 100 = -60%

То фича наоборот, будет увеличивать процент кешируемых блоков до тех пор, пока Overhead не станет положительным.

Ниже будет пример как это выглядит на реальных данных. Не нужно пытаться достигнуть 0%, это невозможно. Весьма хорошо когда когда около 30 — 100%, это помогает избежать преждевременного выхода из режима оптимизации при краткосрочных всплесках.

hbase.lru.cache.heavy.eviction.overhead.coefficient — устанавливает, как быстро мы хотели бы получить результат. Если мы твердо знаем, что наши чтения в основном длительные и не хотим ждать, мы можем увеличить этот коэффициент и получить высокую производительность быстрее.

Например, мы установили этот коэффициент = 0.01. Это означает что Overhead (см. выше) будет умножен на это число на полученный результат и будет уменьшен процент кешируемых блоков. Допустим, что Overhead = 300%, а коэффициент = 0.01, то процент кешируемых блоков будет уменьшен на 3%.

Подобная логика «Backpressure» — реализована и для отрицательных значений Overhead (overshooting). Так как всегда возможны краткосрочные колебания объема чтений-выселений, то этот механизм позволяет избегать преждевременный выход из режима оптимизации. Backpressure имеет перевернутую логику: чем сильнее overshooting, тем тем больше кешируется блоков.




Рассмотрим теперь все это на реальном примере. Имеем следующий тестовый сценарий:

1. Начинаем делать Scan (25 threads, batch = 100)
2. Через 5 минут добавляем multi-gets (25 threads, batch = 100)
3. Через 5 минут выключаем multi-gets (остается опять только scan)

Делаем два прогона, сначала hbase.lru.cache.heavy.eviction.count.limit = 10000 (что фактически выключает фичу), а затем ставим limit = 0 (включает).

В логах ниже мы видим, как включается фича, сбрасывает Overshooting до 14-71%. Время от времени нагрузка снижается, что включает Backpressure и HBase вновь кеширует больше блоков.


Сканы нужны были для того, чтобы показать этот же процесс в виде графика соотношения между двумя разделами кеша — single (куда попадают блоки которые еще никто ни разу не запрашивал) и multi (тут хранятся «востребованные» хотя бы раз данные):


Тут видно как работает механизм корректировки процента кешируемых блоков.

Полный код можно найти в Pull Request HBASE 23887 на github.

Однако 300 тыс. чтений в секунду это не все, что можно выжать на данном железе в этих условиях. Дело в том, что когда нужно обратиться к данным через HDFS, то используется механизм ShortCircuitCache (далее SSC), который позволяет получить доступ к данным напрямую, избегая сетевых взаимодействий.

Профилировка показала, что этот механизм хоть и дает большой выигрыш, но сам также в какой-то момент становится узким горлышком, потому что практически все тяжелые операции происходят внутри lock, что приводит к блокировкам большую часть времени.



Осознав это мы поняли, что проблему можно обойти, если создать массив независимых SSC:

private final ShortCircuitCache[] shortCircuitCache;
...
shortCircuitCache = new ShortCircuitCache[this.clientShortCircuitNum];
for (int i = 0; i < this.clientShortCircuitNum; i++)
  this.shortCircuitCache[i] = new ShortCircuitCache(…);

И далее работать с ними, исключая пересечения так же по последней цифре оффсета:

public ShortCircuitCache getShortCircuitCache(long idx) {
    return shortCircuitCache[(int) (idx % clientShortCircuitNum)];
}

Теперь можно приступать к тестам. Для этого будем читать файлы из HDFS простым многопоточным приложением. Выставляем параметры:

conf.set("dfs.client.read.shortcircuit", "true");
conf.set("dfs.client.read.shortcircuit.buffer.size", "65536"); // по дефолту = 1 МБ и это сильно замедляет чтение, поэтому лучше привести в соответствие к реальным нуждам
conf.set("dfs.client.short.circuit.num", num); // от 1 до 10

И просто читаем файлы:

FSDataInputStream in = fileSystem.open(path);
for (int i = 0; i < count; i++) {
    position += 65536;
    if (position > 900000000)
        position = 0L;
    int res = in.read(position, byteBuffer, 0, 65536);
}

Этот код выполняется в отдельных потоках и мы будем наращивать количество одновременно читаемых файлов (от 10 до 200 — горизонтальная ось) и количество кэшей (от 1 до 10 — графики). Вертикальная оси показывает ускорение которое дает увеличение SSC относительно случая когда кеш только один.



Как читать график: время выполнения 100 тысяч чтений блоками по 64 КБ с одним кэшом требует 78 секунд. Тогда как с 5 кэшами это выполняется за 16 секунд. Т.е. имеет место ускорение ~5 раз. Как видно из графика, на маленьком числе параллельных чтений эффект не очень заметный, это начинает играть заметную роль когда чтения потоков больше 50. Также заметно, что увеличение количества SSC от 6 и выше дает существенно меньше прироста производительности.

Примечание 1: так как результаты тестирования достаточно волатильны (см. ниже), было осуществлено 3 запуска и полученные значения были усреднены.

Примечание 2: Прирост производительности от настройки для случайного доступа такой же, хотя сам доступ чуть медленнее.

Однако необходимо уточнить, что в отличие от случая с HBase это ускорение не всегда бесплатное. Тут мы больше «разблокируем» возможности CPU делать работу, вместо того чтобы отвисать на локах.



Тут можно наблюдать, что в целом увеличение количества кэшей дает примерно пропорциональный рост утилизации ЦПУ. Однако есть несколько более выигрышные комбинации.

Например присмотримся внимательнее к настройке SSC = 3. Рост производительности на диапазоне составляет около 3.3 раз. Ниже результаты всех трех отдельных запусков.



Тогда как потребление CPU растет примерно в 2.8 раз. Разница не очень большая, но маленькой Грете уже радость и возможно появится время для посещения школы и уроков.

Таким образом это будет иметь позитивный эффект для любого инструмента использующего массовый доступ к HDFS (например Spark и т.д.), при условии что прикладной код легкий (т.е. затык именно на стороне клиента HDFS) и есть свободные мощности CPU. Для проверки давайте протестируем какой эффект даст совместное применение оптимизации BlockCache и тюнинга SSC для чтения из HBase.



Тут видно, что в таких условиях эффект не такой большой, как в рафинированных тестах (чтение без всякой обработки), однако выжать дополнительные 80К тут вполне себе получается. Совместно обе оптимизации дают ускорение до 4х раз.

Так же по этой оптимизации был сделан PR [HDFS-15202], который был вмержен и данный функционал будет доступен в следующих релизах.

Ну и наконец было интересно сравнить производительность чтения подобной wide-column БД Cassandra и HBase.

Для этого запускались экземпляры стандартной утилиты нагрузочного тестирования YCSB с двух хостов (800 threads суммарно). На серверной стороне — по 4 экземпляра RegionServer и Cassandra на 4 хостах (не тех, где запущены клиенты, чтобы избежать их влияния). Чтения шли из таблиц размером:

HBase — 300 GB on HDFS (100 GB чистых данных)

Cassandra — 250 GB (replication factor = 3)

Т.е. объем был примерно одинаковый (в HBase немножко больше).

Параметры HBase:

dfs.client.short.circuit.num = 5 (оптимизация клиента HDFS)

hbase.lru.cache.heavy.eviction.count.limit = 30 — это означает то патч начнет работать через 30 выселений (~5 минут)

hbase.lru.cache.heavy.eviction.mb.size.limit = 300 — целевой объем кеширования и выселения

Логи YCSB были распарсены и сведены в графики Excel:



Как видно, данные оптимизации позволяют сравнять производительность этих БД в этих условиях и достигнуть 450 тыс. чтений в секунду.

Надеемся эта информация может быть кому-нибудь полезной в ходе увлекательной борьбы за производительность.