Multiple threads and logstash aggregation filter

Question

I have a pipeline from an SQL database to Elasticsearch that looks something like the following:

1. Input from the SQL database using logstash-input-jdbc
2. Various filtering and mutation of the individual events
3. The events are aggregated according to a group_id property using logstash-filter-aggregate
4. The aggregate events are output to Elasticsearch using logstash-output-elasticsearch

As it is, the throughput of this pipeline is quite low. I know that this is due to the aggregation step (which performs some relatively heavy processing), and I would like to use several threads/processes in order to improve performance (allowing me to utilize more than one core).

However, the logstash-filter-aggregate plugin does not support multiple filter workers -- presumably because it has no way to guarantee that the events that should be combined into one aggregate events will be processed by the same worker.

My current solution to this is to run several instances of logstash where each instance selects a certain subset of group_ids from the SQL database. However, there is quite a bit of overhead to this. Are there any better ways to use multiple cores with logstash-filter-aggregate?

Answer 1

You're in a bit of a hole, there. Aggregate is one of the filters that requires serialization of the event stream, because it considers needful state to be one or more events. Any time you require serialization, your throughput is going to be limited to a single core to ensure the filter worker sees all the events it needs. As with databases, the way around this is to shard your dataset. Which you've already discovered.

Really, sharding is the best single-stage logstash solution to this problem.

There is a multi-stage solution, if you want to go there. That is to build a second pipeline. It would work like this:

1. The first pipeline ingests events and marks up related events as related, but performs no aggregation.
2. First pipeline outputs to elasticsearch as you have been.
3. Second pipeline uses the elasticsearch input to query for events that look like they are un-aggregated events.
4. The aggregation layer aggregates events that are complete (have all of their parts)
5. The aggregation events are flushed to Elasticsearch.

Alternately, you may use non-logstash methods to perform the aggregation once it is in ElasticSearch.