Несколько потоков и агрегации фильтр logstash

Question

У меня есть трубопровод из базы данных SQL к Elasticsearch, который выглядит примерно следующим:

1. вход из базы данных SQL, используя logstash-вход-JDBC
2. Различные фильтрации и мутации отдельных событий
3. события агрегируются в соответствии с gROUP_ID собственности с помощью logstash фильтр-заполнитель
4. совокупные события выводятся на Elasticsearch использованием logstash-выход-elasticsearch

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

Однако плагин logstash-filter-aggregate не поддерживает нескольких работников фильтров - по-видимому, потому, что он не может гарантировать, что события, которые должны быть объединены в одни агрегированные события, будут обрабатываться одним и тем же работником.

Мое настоящее решение состоит в том, чтобы запустить несколько экземпляров logstash, где каждый экземпляр выбирает определенное подмножество group_ids из базы данных SQL. Тем не менее, на это слишком много накладных расходов. Существуют ли какие-либо более эффективные способы использования нескольких ядер с помощью агрегата logstash-filter-aggregate?

Answer 1

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

Действительно, оштрафование является лучшим одноступенчатым решением для решения этой проблемы.

Существует многоэтапное решение, если вы хотите туда попасть. То есть построить второй трубопровод. Он будет работать следующим образом:

1. Первый трубопровод поглощает события и маркирует связанные события как связанные, но не выполняет агрегацию.
2. Первые выходы трубопровода до elasticsearch, как вы были.
3. Второй трубопровод использует вход elasticsearch для запроса событий, которые выглядят как неагрегированные события.
4. Агрегационный агрегат объединяет события, которые являются полными (имеют все их части)
5. События агрегации очищаются до результатов поиска Elasticsearch.

В качестве альтернативы, вы можете использовать методы без логстаста для выполнения агрегации, как только он находится в ElasticSearch.