[Реальный бой] Обмен данными об улучшении скорости записи в elasticsearch

Статья впервые опубликована и отправлена в публичный аккаунт InfoQ [заметки о разработке Сямэна], добро пожаловать на внимание.
- https://www.infoq.cn/article/t7b52mbzxqkwrrdpVqD2

базовая конфигурация

Базовая конфигурация, 5 хостов, сконфигурированных как 24C 125G 17T, и на каждом хосте построен узел elasticsearch.
Используемая версия кластера elasticsearch — 7.1.1. Инструменты управления включают kibana и cerebro.

Приложения

Источниками данных являются три темы кафки, которые в основном используются для хранения и извлечения данных журнала в реальном времени.Из-за требований в реальном времени необходимо быстро записывать данные в es.
Назовем их соответственно TopicA, TopicB, TopicC. Из-за настройки записи некоторые основные показатели исходных данных необходимо детально разобрать, чтобы облегчить последующий анализ. Ниже приводится генерация данных по трем темам:

file

планирование

Topic	Размер пакета	Расчетный объем данных (дни)	Планирование количества осколков индекса es
topicA	900b	2T	30 основных фрагментов
topicB	800b	400G	10 основных фрагментов
topicC	750b	300G	10 основных осколков

Проблема воспроизводится

Без какой-либо настройки, используя java и logstash для извлечения данных соответственно, эффективность невелика.Конкретные проблемы заключаются в следующем:
- 1. Существует серьезное отставание данных Kafka, и потребление не поспевает за скоростью производства.
- 2. Кластер elasticsearch находится под высокой нагрузкой, и большое количество операций записи отклоняется.
- 3. Java-программа часто генерирует исключение RejectionException.
- 4. Ненормально высокая нагрузка на центральный процессор.
  Настройка конфигурации на уровне операционной системы и JVM здесь описываться не будет, а об этом типе есть много статей, на которые можно сослаться.
  Настраиваем каждую деталь по модулю, детали таковы:

Оптимизация записи

От количественного к количественному

В используемой java-программе мы изменили фиксированное количество вставок на вставки фиксированного размера.Из-за используемой более высокой версии es он был напрямую заменен официально рекомендованным методом BulkProcessor. Указанные свойства следующие:

# 每2w条执行一次bulk插入
bulkActions: 20000
# 数据量达到15M后执行bulk插入
 bulkSizeMb: 15
# 无论数据量多少，间隔20s执行一次bulk
flushInterval: 20
# 允许并发的bulk请求数
concurrentRequests: 10

Конкретные значения конфигурации здесь можно постепенно увеличивать в соответствии с наблюдаемым состоянием кластера. Для высоких версий es вы можете наблюдать за скоростью индексирования на странице мониторинга x-pack и соответствующим образом настраивать.Если версия es низкая, вы можете использовать рекомендуемый rest API для логической инкапсуляции.
В младшей версии es идея определения скорости записи заключается в следующем: написать программу для регулярной проверки объема данных индекса для расчета. Если вы используете python, вы можете получить общий объем данных, проиндексированных в две строки кода.

 call_list = es.indices.stats(index=index)
total = call_list['indices'][index]['total']['indexing']['index_total']。

Вы также можете использовать CURL для приблизительного подсчета размера данных одного индекса каждые несколько минут. Команда выглядит следующим образом:

Запрос общего количества проиндексированных документов

curl -XGET -uname:pwd
'http://esip:port/_cat/count/index-name?v&format=json&pretty'

запустить несколько процессов

Поскольку Bulkprocess является потокобезопасным, мы можем использовать несколько потоков для совместного использования пакетного процессора. Лучший способ потребления — запустить несколько процессов-потребителей, развернуть их на разных хостах, сделать общее количество многопотоков, открытых в нескольких процессах, равным количеству разделов темы, и установить их как одну и ту же группу потребителей. Каждый процесс содержит объемный процесс, который может улучшить возможности потребления и пакетной записи. В то же время можно избежать одноточечных проблем: если процесс-потребитель зависнет, кластер kafka проведет перебалансировку разделенных потребителей. Отсутствие потребителей повлияет только на скорость потребления, но не на обработку данных.

«Испытание давлением» для увеличения количества вставок в партиях

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

1. Наблюдайте за значением Load Average нагрузки кластера.

Значение нагрузки в определенной степени отражает загруженность ЦП, так как же нам интерпретировать значение нагрузки на странице мониторинга elasticsearch? Ниже приведен кластер из трех узлов. На интерфейсе, предоставленном cerebo слева, значение нагрузки отмечено красным, что указывает на то, что нагрузка es может быть немного высокой, поэтому какое значение будет отображаться красным? Давайте учиться вместе. .

file file

Начиная с уровня хоста, под Linux предоставляется команда uptime для наблюдения за нагрузкой хоста.
Три значения средней нагрузки представляют ситуацию загрузки хоста за 1 минуту, 5 минут и 15 минут соответственно. Некоторые люди могут задаться вопросом, означает ли 26.01, что загрузка хоста составляет 26%?С кластера es, который мы запускаем, это явно не низкая производительность нагрузки. На самом деле, в случае одного процессора это значение можно рассматривать как процент, например, загрузка 0,05 означает, что текущая загрузка системы составляет 5%. Но наши серверы, как правило, имеют несколько процессоров, и каждый процессор будет содержать несколько ядер ЦП, поэтому отображаемое здесь значение связано с количеством ядер ЦП.Если вам нужно использовать процент для представления загрузки системы, вы можете разделить конкретную нагрузку значение на общее количество ядер сервера, чтобы увидеть, больше ли оно 1. Команда для просмотра общего количества ядер:

cat /proc/cpuinfo |grep -c 'model name'

Этот хост отображается как 24. Из нагрузки 26 текущие обрабатываемые задачи должны быть поставлены в очередь, поэтому загрузка отмечена красным. В то же время, вот список того, как проверить состояние процессора

总逻辑CPU数 = 物理CPU个数 X 每颗物理CPU的核数 X 超线程数
# 物理CPU个数（我们的服务器是2个）
cat /proc/cpuinfo| grep "physical id"| sort| uniq| wc -l


# 查看每个物理CPU中core的个数(就是核数)（6核）
cat /proc/cpuinfo| grep "cpu cores"| uniq

# 查看逻辑CPU的个数
cat /proc/cpuinfo| grep "processor"| wc -l
（显示24，不等于上面的cpu个数 * 每个cpu的核数，说明是开启了超线程）

2. Наблюдаем, чем занят кластер

Через таски апи можно интуитивно наблюдать чем занят кластер? , включая такие показатели, как родительская задача, продолжительность задачи и т. д. Команда выглядит следующим образом:

curl -u username:pwd ip:port/_cat/tasks/?v | more

file

Выше приведен снимок экрана, который я сделал после установки копии на 0. Теоретически также должна быть операция bulk[s][r]. Видно, что запись сейчас очень трудоемка, в норме пакет массовых операций должен быть в миллисекундах, что также показывает, что нагрузка на es очень велика.
из задачиидентификатор, родительКак видно из task_id, массовая операция делится на действие записи основного шарда и действие записи реплики. в:
index:data/write/bulk[s][p]: s представляет осколок, а p представляет основной осколок.
index:data/write/bulk[s][r]: s означает сегмент, r означает копирование.

Простой процесс написания

Ниже приведен простой процесс написания массового запроса. Мы знаем, что клиент выберет узел для отправки запроса. Этот узел называется узлом-координатором, также известным как клиентский узел, но перед выполнением, если операция предварительной обработки конвейера определено (например, значение ключа преобразуется перед записью или добавляется метка времени и т. д.), операция записи будет перехвачена и будет выполнена соответствующая логическая обработка.
- Как видно из рисунка, операция записи теперь будет решать, в какой сегмент отправлять данные в соответствии с правилами маршрутизации.По умолчанию маршрутизация выполняется по идентификатору документа данных, что обеспечивает равномерное распределение данных по каждому shard.На узле также можно настроить правила маршрутизации.О конкретном способе определения мы поговорим ниже.
  - Затем запрос отправляется на основной сегмент.После успешного выполнения основного сегмента запрос будет перенаправлен на соответствующий сегмент реплики.После успешного выполнения на сегменте реплики запрос считается выполненным, а затем результат выполнения возвращается клиенту end.
  - Видно, что в сценарии больше писать и меньше читать многократные копии потребляют много производительности, примерно несколько дополнительных копий эквивалентны многократной записи нескольких копий данных. Если отказоустойчивость данных не учитывается, количество реплик может быть соответствующим образом уменьшено или реплики могут быть удалены для повышения скорости записи.
    Узел с ролью приема не используется в нашем кластере, он предлагается здесь, чтобы помочь всем лучше понять разделение ролей каждого узла.

file

    - 通过ES提供的API观察各个节点的热线程，api结果会显示出占用cpu高的线程，这也是我们可以优化的地方。大量写入场景下，这里一般大多数会显示：Lucene Merge Thread 或者[write]，查询命令为：
GET /_nodes/hot_threads

3. Наблюдайте за состоянием пула потоков кластера.

Чтобы избежать отклонения большого количества пишетов, вы можете соблюдать использование внутреннего пула резьбы и соответствующего размера очереди, наблюдая за журналом фона ELASTICSearch или с помощью API пула резьбы, чтобы определить, сможете ли вы продолжать настроить параметры конфигурации записи Отказ

curl -uusername:pwd-XGET "http://esip:port/_cat/thread_pool?v" | grep write

При высокой нагрузке на запись может произойти большое количество отклонений, например:

node-name	name	active	queue
node-3	write	4	0
node-1	write	3	2
node-2	write	9	1

основная часть

Каждый каталог монтирует отдельный диск

В каталоге данных мы разделили 10 подкаталогов и смонтировали их на разных жестких дисках. Это эквивалентно выполнению raid0. Может значительно улучшить скорость записи.

Настроить несколько path.data

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

Один хост запускает две ноды

Выделенная память экземпляра ES не будет превышать 32G. Для нас с фиксированным количеством хостов, если только один узел ES установлен на машине 125G, он немного расточился, поэтому рассмотрите начать два экземпляра узла ES на хосте.
При настройке необходимо обратить внимание на следующие моменты:
- 1. Настройки HTTP-порта и транспортного порта для связи между узлами.
- 2. Назначение роли узла.
- 3. Конфигурация с разделенным мозгом изменена соответствующим образом.
- 4, модификация атрибута path.data (важно)
- 5, изменение атрибута path.logs.

Измените конфигурацию path.data таким образом, чтобы два узла одного и того же хоста в равной степени использовали жесткий диск.

Здесь мы сосредоточимся на пункте 4. После запуска двух экземпляров на одном хосте мы изменили конфигурацию path.data с исходных 10 каталогов на 5 разных каталогов для каждой конфигурации.

path.data: /data01/esdata,/data02/esdata,/data03/esdata
,/data04/esdata,/data05/esdata

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

заменить твердотельный накопитель

Твердотельные накопители могут удвоить скорость записи.Если вы используете твердотельные накопители, возможно, вам не придется выбрасывать эту статью.

раздел эластичного поиска

Настройки роли узла

Elasticsearch предлагает несколько типов настроек ролей узла, которые необходимо указать в конфигурации elasticsearch.yml.
тип.

Укажите шаблон индекса

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

{
  "order": 0,
  "index_patterns": [
    "topicA*"
  ],
  "settings": {
    "index": {
      "refresh_interval": "40s",
      "number_of_shards": "30",
      "translog": {
        "flush_threshold_size": "1024mb",
        "sync_interval": "120s",
        "durability": "async"
      },
      "number_of_replicas": "0",
      "merge": {
        "scheduler": {
          "max_thread_count": "1"
        }
      }
    }
  },
  "mappings": {
  },
  "aliases": {}
}

Подсчитать количество осколков

Следует отметить, что количество шардов лучше всего задавать целым числом, кратным количеству узлов, чтобы обеспечить почти одинаковую нагрузку на каждый хост, особенно если на хосте развертывается несколько инстансов, уделяйте этому больше внимания, иначе вы могут столкнуться с другими хостами.Нагрузка нормальная, а нагрузка на определенный хост особенно высока.
Как правило, мы рассчитываем сегменты в соответствии с ежедневным объемом данных, и разумно поддерживать размер каждого сегмента ниже 50G. Если требования не могут быть выполнены, это может быть достигнуто путем разделения большего количества индексов на уровне индекса или путем создания псевдонимов + создания индексов по часам.

Шарты контроля равномерно распределены на каждом хосте

Взяв в качестве примера индекс данных TopicA, всего на 10 узлах выделено 30 сегментов, каждому узлу должно быть выделено 3 сегмента, а всего 6 сегментов на хосте считаются сбалансированными. Если распределение не такое, миграцию шардов можно выполнить с помощью cerebo или через командную строку.

file file

curl -X POST "localhost:9200/_cluster/reroute?pretty" -H 'Content-Type: application/json' -d'
{
    "commands" : [
        {
            "move" : {
                "index" : "test", "shard" : 0,
                "from_node" : "node1", "to_node" : "node2"
            }
        }
    ]
}

Настройте индексный буфер

То есть указать index.memory.indexbufferРазмер размера, это статическая переменная, вам нужно изменить файл конфигурации, чтобы он вступил в силу после перезапуска.
Справочная формула расчета: index.memory.indexbuffersize / shards_count > 512 МБ (более этого значения производительность индекса существенно не улучшится)
Shards_Count - это количество фрагментов над узлом, который можно настроить или процент от общего значения памяти ES. Здесь мы модифицировали 20% (по умолчанию 10%).

file

    
###### 路由分片
- 可以使用elasticsearch提供的routing特性，将数据按一定规则计算后(内部采用hash算法，把相同hash值的文档放入同一个分片中)，默认情况下是使用DocId来计算，写入到分片，查询时指定routing查询，则可以提高查询速度，避免了扫描过多的分片带来的性能开销。首先，在创建索引模板的时候，需要在mappings中增加配置,要求匹配到此索引模板的索引，必须配置routing：

"_routing": {      
    "required": true
  }

На втором шаге при создании IndexRequest для BulkPorcess укажите значение, участвующее в вычислении хэша, через метод routing(java.lang.String routing). Обратите внимание, что это конкретное значение, а не имя поля.

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

file

вытоптанная яма

1. Настройка ролей узла

Если количество нод в кластере невелико, а данные не нуждаются в предварительной обработке, то от использования нод типа Ingest можно вообще отказаться. Параметр по умолчанию для всех узлов по умолчанию равен true. Поэтому мы вручную устанавливаем главный узел и узел данных следующим образом.
node.ingest: false

Но следует знать, что X-Pack отслеживает этот тип узла. Будет неверным: не удалось сбросить массивы экспорта, ни одно решение узла загрузки не должно открывать эту конфигурацию свойства или указано в ElasticSearch.yml: xpack.monitoring.exporters.my_local: Type: xpack.monitoring.exporters.local USE_INGEST: FALSE

2, Elasticsearch Thread Pount Configuration Параметры изменения

Начиная с версии 5.0 запрещено изменять тип пула потоков каждого модуля, а префикс конфигурации, связанной с пулом потоков, изменился с threadpool на threadпул. И уровень конфигурации, связанный с пулом потоков, повышен до конфигурации уровня узла. Запрещено изменять его с помощью API. Поскольку сценарий заключается в том, чтобы писать больше и читать меньше, мы просто увеличиваем размер очереди записи. и настройте его как: потокpool.write.queue_size: 1000.
Его можно изменить, только изменив файл конфигурации.

3. Слишком высокая нагрузка на один хост

Два узла одного и того же хоста являются узлами данных, и сегменты распределяются неравномерно, в результате чего коэффициент использования ЦП на этом хосте составляет около 98% Позже нагрузка снижается за счет переноса сегментов.

4. Пишите о горячих проблемах после использования настраиваемых правил маршрутизации

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

5. Проблема разницы часовых поясов данных в 8 часов

Преобразование рабочего времени в строку с часовым поясом. гггг-ММ-дд'T'ЧЧ:мм:сс.SSSZ

Справочная статья:

http://kane-xie.github.io/2017/09/09/2017-09-09_Elasticsearch%E5%86%99%E5%85%A5%E9%80%9F%E5%BA%A6%E4%BC%98%E5%8C%96/https://www.elastic.co/guide/en/elasticsearch/reference/5.0/breaking50settings_changes.htmlhttps://elasticsearch.cn/question/1915https://juejin.im/entry/6844903873153335309

Добро пожаловать в официальный аккаунт [заметки Сямэна о разработке], чтобы общаться и развиваться вместе