Elasticsearch-rails: Bulk index and Sidekiq: java.lang.OutOfMemoryError: Java heap space

[phlegx]

Hi!

I use the Elasticsearch::Persistence.client client to perform parallel bulk inserts in ES 5.2. The bulk insert is done by sidekiq jobs.

Here some config information:

- bootstrap.memory_lock=true
- ES_JAVA_OPTS=-Xms5g -Xmx5g
- maximal memory limited to 10 G
- thread_pool.bulk.queue_size: 500
- CPU 8 cores
- Sidekiq processes: 6
- Sidekiq threads per process: 15
- All processes has all 15 threads processing an IndexerJob
- All done proposed here: https://www.elastic.co/guide/en/elasticsearch/reference/current/tune-for-indexing-speed.html

Here a small example:

class IndexerJob < ActiveJob::Base
  def perform(objects)
    data = # Some Ruby process with DB queries
    Elasticsearch::Persistence.client.
      bulk(index: MyModel.index_name,
              type: MyModel.document_type,
              body: data)
  end
end

I get this error in ES after some jobs are processed:

java.lang.OutOfMemoryError: Java heap space
Dumping heap to java_pid1.hprof ...
Heap dump file created [5474370005 bytes in 45.184 secs]

Question: Do I need to flush the bulk insert after insert to reduce heap memory or what I am doing wrong?

Another question: Are the bulk inserts asynchronous or not?

I have read something about bulk here: https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/java-docs-bulk-processor.html (setFlushInterval, setConcurrentRequests). Can I config the bulk insert with elasticsearch-rails?

[karmi]

Hi, first, thanks for all the detailed info!

Do I need to flush the bulk insert after insert to reduce heap memory or what I am doing wrong?

in general, it is best to leave Elasticsearch at defaults. That said, you might experiment here by executing the indices.flush_synced operation between the batches, and observe the effect on the memory. Also, if this is a heavy import, which takes significant time, you might want to experiment with disabling the refresh interval to see if it has any desirable effect.

Are the bulk inserts asynchronous or not?

Not on the Ruby layer, and they are not "asynchronous", but parallel on the Elasticsearch layer -- the requests for each shard are executed on corresponding node in parallel.

I've sometimes experiment with making the operations asynchronous on the Ruby layer (using process/threads), and I always felt like the throughput increase isn't worth it. That said, I'd like to experiment by using https://github.com/grosser/parallel and capture some benchmarks, if there'd be more time.

That said, you probably have a _lot_ of asynchronicity coming from the fact that you're using Sidekiq for processing the operations -- essentially pushing data to Elasticsearch in parallel, in 15 threads. Or, and I'd need to verify this, in 6 * 15 threads (15 threads per worker, 6 workers in total). This might easily overload the cluster, because it cannot keep up. Are you checking for the 429 HTTP status code coming from Elasticsearch, and stalling the jobs in that case a bit, effectively applying this back-pressure to the Sidekiq layer?

(Mental note to myself: this should be added to the example in the Rails example application.)

I have read something about bulk here:

These parameters apply to the Java API, not the HTTP REST API, which the Ruby client uses.

Regarding your main question about the out-of-memory error, my best bet is that you're overloading the cluster either with too much parallelism or too big batches. So, the easiest way is to cut down the number of Sidekiq workers, and see what happens. If that helps, you can go in the other direction, and decrease the number of documents (objects in your code) you're sending in each bulk operation, and see what happens. Depending on your data and the desired requirements, you might find a good balance here.

Of course, there knobs to turn, such as indices.memory.index_buffer_size, or other strategies, like temporarily increasing the number of nodes in the cluster, but I'd first try the techniques applied in the former paragraph.

Can you check these out and let me know how everything behaves? Heavy bulk loading is an important use case and the more info there is out there, the better!

[phlegx]

Thank you for the detailed answer. My seed pre-configures ES with:

'index.refresh_interval' => -1,
'index.number_of_replicas' => 0

Sidekiq

I have some Ruby processing and DB queries in my IndexerJob before the bulk import happens. An important note is, that Sidekiq threads (in my case 15 threads per worker process) are concurrent but not parallel. Only the 6 worker processes works in parallel. And I don't get 429 HTTP code errors.

Index buffer size

I have found indices.memory.index_buffer_size on ES documentation. Default is 10% of the JVM heap memory. Actually I have running the seed with the following config without crash:

HW config

• CPU 8 cores
• about 40 GB RAM
• normal server HD 7200 rpm (not SSD)
• ulimits hard and soft 65536
• vm.max_map_count=262144

ES config

• bootstrap.memory_lock=true
• indices.memory.index_buffer_size=10%
• thread_pool.bulk.queue_size=500
• ES_JAVA_OPTS=-Xms10g -Xmx10g
• maximal memory limitation to 20 G

ES index config

• number_of_shards=5
• number_of_replicas=1

ES transient config (only on seed)

• index.refresh_interval=-1
• index.number_of_replicas=0

Sidekiq config

• Sidekiq processes: 6
• Sidekiq threads per process: 15
• All processes has all 15 threads processing an IndexerJob
• The bulk document size is not constant. It varies from 4000 to 5500 documents per bulk insert.

Any idea why the Index Memory is only rising (linear)?

[karmi]

Thanks for all the details!

Actually I have running the seed with the following config without crash (...)

So, what has changed since your original configuration?

Any idea why the Index Memory is only rising (linear)?

Yes, that seems quite logical to me -- when you look at the _type_ of the memory, it's the "Terms" part of the "Index Memory" output. It grows with the amount of documents you keep adding to the index during the bulk load, because that's the fundamental Lucene data structure keeping the inverted index.

This should correspond to the curl 'http://localhost:9200/_cluster/stats?pretty&human' | jq '.indices.segments.terms_memory' part of the "Cluster Stats" API output.

If you would be interested in the lowest-level details, you may check the "Segments" API in verbose mode: curl 'http://localhost:9200/_segments?human=true&verbose=true' | jq ., which would give you the breakdown for specific fields. (Not sure now if it's available on 2.x though.)

[karmi]

Actually I have running the seed with the following config without crash (...)
So, what has changed since your original configuration?

I'm still interested if anything has changed in your configuration and/or environment, since the operation has been failing previously, and now runs fine. Any news there? :)

[phlegx]

Hi! The problem was the amount of RAM. Giving more RAM to the machine and to the JVM solves the problem. In fact, the Index Memory has reached the maximal JVM heap memory. That causes the error. Working with the following memory configuration all works fine for the moment:

• ES_JAVA_OPTS=-Xms20g -Xmx20g
• maximal memory limitation to 40 G

[phlegx]

Here a good documentation that I have found: https://thoughts.t37.net/how-we-reindexed-36-billions-documents-in-5-days-within-the-same-elasticsearch-cluster-cd9c054d1db8

[estolfo]

Closing because this has been resolved.