# BulkProcessor usage is safe?
**URL:** https://discuss.elastic.co/t/bulkprocessor-usage-is-safe/12625
**Category:** Elasticsearch
**Created:** [July 2, 2013, 7:04pm UTC](https://discuss.elastic.co/t/bulkprocessor-usage-is-safe/12625 "2013-07-02T19:04:15Z")
**Posts on this page:** 7
**Page:** 1
### Author:  [@Thomas\_Bolis](https://discuss.elastic.co/u/Thomas_Bolis)
#### Post date: [July 2, 2013, 7:04pm UTC](https://discuss.elastic.co/t/bulkprocessor-usage-is-safe/12625/1 "2013-07-02T19:04:15Z")
Hi,
I'm trying to implement a solution for fast bulk insert of data into ES.
I'm currently noticing that it is the fastest way of inserting data. I have
also tried to implement it through a multi threaded approach with
BulkRequestBuilder but I wasn't able to reach the throughput of
BulkProcessor.
My problem is that at some point I receive exceptions in Listener like _no
node is available_ and the most important is that if I stress it much i see
more docs than the ones inserted (e.g. i insert 500000 docs and i finally
see something like 518018 docs)
Your help will be much appreciated.
I create a processor as follows:
----- CODE ------
processor = BulkProcessor.builder(client, new BulkProcessor.Listener() {
public void beforeBulk(long executionId, BulkRequest
request) {
log.debug(String.format("Execution: %s, about execute
new bulk insert composed of {%s} actions", executionId,
request.numberOfActions()));
}
```
public void afterBulk(long executionId, BulkRequest
```
request, BulkResponse response) {
log.debug(String.format("Execution: %s, bulk insert
composed of {%s} actions, took %s", executionId,
request.numberOfActions(),
formatUtils.getDurationDHMS(response.getTookInMillis())));
}
```
public void afterBulk(long executionId, BulkRequest
```
request, Throwable failure) {
// throw new RuntimeException(String.format("Execution
with id %s failed", executionId), failure);
}
}).setConcurrentRequests(500).build();
----- CODE ------
--
You received this message because you are subscribed to the Google Groups "elasticsearch" group.
To unsubscribe from this group and stop receiving emails from it, send an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).
For more options, visit [https://groups.google.com/groups/opt\_out](https://groups.google.com/groups/opt_out).
---
### Author:  [@jprante](https://discuss.elastic.co/u/jprante)
#### Post date: [July 2, 2013, 8:27pm UTC](https://discuss.elastic.co/t/bulkprocessor-usage-is-safe/12625/2 "2013-07-02T20:27:29Z")
Examine your code (which I can't find in your post) where you add items
to BulkRequest. This procedure must be thread safe. Next, check your doc
ID creation if it is really unique.
Note, if your code snippet is really true, 500 concurrent requests will
massively slow down your bulk ingestion and add a significant overhead
to thread management (and, by default, the ES pools are not this large
that you can really expect this to work efficiently).
"No node available" is no surprise under this circumstances, it is an
exception if you "overwhelm" ES and it can't respond within 5 seconds.
You need to streamline your bulk ingest - beside less concurrency, check
how much throughput your node heap & disk i/o can process, and adjust
the client bulk max actions and max concurrency.
Jörg
Am 02.07.13 21:04, schrieb Thomas Bolis:
> Hi,
>
> I'm trying to implement a solution for fast bulk insert of data into
> ES. I'm currently noticing that it is the fastest way of inserting
> data. I have also tried to implement it through a multi threaded
> approach with BulkRequestBuilder but I wasn't able to reach the
> throughput of BulkProcessor.
>
> My problem is that at some point I receive exceptions in Listener like
> /no node is available/ and the most important is that if I stress it
> much i see more docs than the ones inserted (e.g. i insert 500000 docs
> and i finally see something like 518018 docs)
>
> Your help will be much appreciated.
>
> I create a processor as follows:
>
> ----- CODE ------
> processor = BulkProcessor.builder(client, new BulkProcessor.Listener() {
> public void beforeBulk(long executionId, BulkRequest
> request) {
> log.debug(String.format("Execution: %s, about
> execute new bulk insert composed of {%s} actions", executionId,
> request.numberOfActions()));
> }
>
> ```
> public void afterBulk(long executionId, BulkRequest
>
> ```
>
> request, BulkResponse response) {
> log.debug(String.format("Execution: %s, bulk
> insert composed of {%s} actions, took %s", executionId,
> request.numberOfActions(),
> formatUtils.getDurationDHMS(response.getTookInMillis())));
> }
>
> ```
> public void afterBulk(long executionId, BulkRequest
>
> ```
>
> request, Throwable failure) {
> // throw new
> RuntimeException(String.format("Execution with id %s failed",
> executionId), failure);
> }
> }).setConcurrentRequests(500).build();
> ----- CODE ------
>
> --
> You received this message because you are subscribed to the Google
> Groups "elasticsearch" group.
> To unsubscribe from this group and stop receiving emails from it, send
> an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).
> For more options, visit [https://groups.google.com/groups/opt\_out](https://groups.google.com/groups/opt_out).
--
You received this message because you are subscribed to the Google Groups "elasticsearch" group.
To unsubscribe from this group and stop receiving emails from it, send an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).
For more options, visit [https://groups.google.com/groups/opt\_out](https://groups.google.com/groups/opt_out).
---
### Author:  [@brian\_yoder](https://discuss.elastic.co/u/brian_yoder)
#### Post date: [July 3, 2013, 12:20am UTC](https://discuss.elastic.co/t/bulkprocessor-usage-is-safe/12625/3 "2013-07-03T00:20:58Z")
Thomas,
See [https://groups.google.com/d/msg/elasticsearch/jUoI1xGAVio/dnY11\_VUL4AJ](https://groups.google.com/d/msg/elasticsearch/jUoI1xGAVio/dnY11_VUL4AJ)
for my post to someone else about how to use the BulkRequestBuilder. It's
not fully filled-out code, but it is nice high-level pseudocode that should
help a lot.
One other thing: Elasticsearch threading, especially in 0.90 and above,
does such a superb job that my bulk loader is serial (one thread). Besides,
my customer's updates contain a mix of deletes and inserts, and they must
be processed in a specific order. For example, if a delete for A is
followed by an insert for A, it would be a very bad thing to multithread
this and have one thread issue the insert for A before another thread
issued the delete for A. So what my customer gives me is also the one of
the best ways to use Elasticsearch.
Hope this helps!
Brian
On Tuesday, July 2, 2013 3:04:15 PM UTC-4, Thomas Bolis wrote:
> Hi,
>
> I'm trying to implement a solution for fast bulk insert of data into ES.
> I'm currently noticing that it is the fastest way of inserting data. I have
> also tried to implement it through a multi threaded approach with
> BulkRequestBuilder but I wasn't able to reach the throughput of
> BulkProcessor.
>
> My problem is that at some point I receive exceptions in Listener like _no
> node is available_ and the most important is that if I stress it much i
> see more docs than the ones inserted (e.g. i insert 500000 docs and i
> finally see something like 518018 docs)
>
> Your help will be much appreciated.
>
> I create a processor as follows:
>
> ----- CODE ------
> processor = BulkProcessor.builder(client, new BulkProcessor.Listener() {
> public void beforeBulk(long executionId, BulkRequest
> request) {
> log.debug(String.format("Execution: %s, about execute
> new bulk insert composed of {%s} actions", executionId,
> request.numberOfActions()));
> }
>
> ```
> public void afterBulk(long executionId, BulkRequest
>
> ```
>
> request, BulkResponse response) {
> log.debug(String.format("Execution: %s, bulk insert
> composed of {%s} actions, took %s", executionId,
> request.numberOfActions(),
> formatUtils.getDurationDHMS(response.getTookInMillis())));
> }
>
> ```
> public void afterBulk(long executionId, BulkRequest
>
> ```
>
> request, Throwable failure) {
> // throw new RuntimeException(String.format("Execution
> with id %s failed", executionId), failure);
> }
> }).setConcurrentRequests(500).build();
> ----- CODE ------
--
You received this message because you are subscribed to the Google Groups "elasticsearch" group.
To unsubscribe from this group and stop receiving emails from it, send an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).
For more options, visit [https://groups.google.com/groups/opt\_out](https://groups.google.com/groups/opt_out).
---
### Author:  [@Thomas\_Bolis](https://discuss.elastic.co/u/Thomas_Bolis)
#### Post date: [July 3, 2013, 7:14am UTC](https://discuss.elastic.co/t/bulkprocessor-usage-is-safe/12625/4 "2013-07-03T07:14:58Z")
First of all Thank you all for your reply :),
Dear Jorg, I will try it with less concurrent requests, but the thing is
that performance will drop probably I will go to my other implementation
which is through BulkRequestBuilder which I don't have these issues.
My question is though whether BulkProcessor is a class meant to be used
generally or it is an internal class for elasticsearch engine an I probably
do not use it well..
Dear Brian, between the implementation proposed and mine there is one
difference the bulkRequest.setRefresh(false). But i check the code and it
is false by default. Probably because i use 0.90.2 and maybe on version
0.20 wasn't the case.
I will try again your proposals and let you know.
Thanks again
Thomas
On Wed, Jul 3, 2013 at 3:20 AM, InquiringMind [brian.from.fl@gmail.com](mailto:brian.from.fl@gmail.com)wrote:
> Thomas,
>
> See [https://groups.google.com/d/msg/elasticsearch/jUoI1xGAVio/dnY11\_VUL4AJfor](https://groups.google.com/d/msg/elasticsearch/jUoI1xGAVio/dnY11_VUL4AJfor) my post to someone else about how to use the BulkRequestBuilder. It's
> not fully filled-out code, but it is nice high-level pseudocode that should
> help a lot.
>
> One other thing: Elasticsearch threading, especially in 0.90 and above,
> does such a superb job that my bulk loader is serial (one thread). Besides,
> my customer's updates contain a mix of deletes and inserts, and they must
> be processed in a specific order. For example, if a delete for A is
> followed by an insert for A, it would be a very bad thing to multithread
> this and have one thread issue the insert for A before another thread
> issued the delete for A. So what my customer gives me is also the one of
> the best ways to use Elasticsearch.
>
> Hope this helps!
>
> Brian
>
> On Tuesday, July 2, 2013 3:04:15 PM UTC-4, Thomas Bolis wrote:
>
> > Hi,
> >
> > I'm trying to implement a solution for fast bulk insert of data into ES.
> > I'm currently noticing that it is the fastest way of inserting data. I have
> > also tried to implement it through a multi threaded approach with
> > BulkRequestBuilder but I wasn't able to reach the throughput of
> > BulkProcessor.
> >
> > My problem is that at some point I receive exceptions in Listener like _no
> > node is available_ and the most important is that if I stress it much i
> > see more docs than the ones inserted (e.g. i insert 500000 docs and i
> > finally see something like 518018 docs)
> >
> > Your help will be much appreciated.
> >
> > I create a processor as follows:
> >
> > ----- CODE ------
> > processor = BulkProcessor.builder(client, new BulkProcessor.Listener() {
> > public void beforeBulk(long executionId, BulkRequest
> > request) {
> > log.debug(String.format("\*\*Execution: %s, about
> > execute new bulk insert composed of {%s} actions", executionId,
> > \*\*
> > request.numberOfActions()));
> > }
> >
> > ```
> > public void afterBulk(long executionId, BulkRequest
> >
> > ```
> >
> > request, BulkResponse response) {
> > log.debug(String.format("\*\*Execution: %s, bulk
> > insert composed of {%s} actions, took %s", executionId,
> > \*\* request.numberOfActions(),
> > formatUtils.getDurationDHMS(\*\*response.getTookInMillis())));
> > }
> >
> > ```
> > public void afterBulk(long executionId, BulkRequest
> >
> > ```
> >
> > request, Throwable failure) {
> > // throw new RuntimeException(String.\*\*format("Execution
> > with id %s failed", executionId), failure);
> > }
> > }).setConcurrentRequests(500).\*\*build();
> > ----- CODE ------
> >
> > --
> > You received this message because you are subscribed to a topic in the
> > Google Groups "elasticsearch" group.
> > To unsubscribe from this topic, visit
> > [https://groups.google.com/d/topic/elasticsearch/vUtJKbjb1bo/unsubscribe](https://groups.google.com/d/topic/elasticsearch/vUtJKbjb1bo/unsubscribe).
> > To unsubscribe from this group and all its topics, send an email to
> > [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).
> > For more options, visit [https://groups.google.com/groups/opt\_out](https://groups.google.com/groups/opt_out).
--
You received this message because you are subscribed to the Google Groups "elasticsearch" group.
To unsubscribe from this group and stop receiving emails from it, send an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).
For more options, visit [https://groups.google.com/groups/opt\_out](https://groups.google.com/groups/opt_out).
---
### Author:  [@jprante](https://discuss.elastic.co/u/jprante)
#### Post date: [July 3, 2013, 11:01am UTC](https://discuss.elastic.co/t/bulkprocessor-usage-is-safe/12625/5 "2013-07-03T11:01:09Z")
Note that ES thread pools are aligned with the number of CPU cores you
have. A higher number of threads will not result in higher performance.
So, if you must really use 500 concurrent threads for best bulk
performance, you have either the world largest CPU core machine, or your
code have some other fundamental problems. You should check if you can
process 8-10 MB bulk data per second per node (for example, monitor the
network connection).
The BulkProcessor class is not marked for internal use (as no ES class
is). I think as long as it is not marked "@deprecated" it will stay in
ES API (if not I could provide substitution class as a plugin because
BulkProcessor does no magic, but I think this will not be necessary)
Remember, the people at ES are really helpful and supportive and high
performance bulk indexing is a strong requirement for many of us in the
community.
Jörg
Am 03.07.13 09:14, schrieb Thomas Bolis:
> Dear Jorg, I will try it with less concurrent requests, but the thing
> is that performance will drop probably I will go to my other
> implementation which is through BulkRequestBuilder which I don't have
> these issues.
>
> My question is though whether BulkProcessor is a class meant to be
> used generally or it is an internal class for elasticsearch engine an
> I probably do not use it well..
--
You received this message because you are subscribed to the Google Groups "elasticsearch" group.
To unsubscribe from this group and stop receiving emails from it, send an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).
For more options, visit [https://groups.google.com/groups/opt\_out](https://groups.google.com/groups/opt_out).
---
### Author:  [@Thomas\_Bolis](https://discuss.elastic.co/u/Thomas_Bolis)
#### Post date: [July 3, 2013, 1:25pm UTC](https://discuss.elastic.co/t/bulkprocessor-usage-is-safe/12625/6 "2013-07-03T13:25:16Z")
Dear Jörg,
Thank you very much for your reply, finally I went with the implementation
of BulkRequestBuilder. The 500 concurrent request is of course a stress I
provide to our cluster, i ended up with bigger queues though. We will now
play a bit with more CPU/Memory. Although I think that CPU is the key to
instantly see performance increase. Still new with Elasticsearch but
excited to dig more into it 🙂
Thanks again
Thomas
On Wed, Jul 3, 2013 at 2:01 PM, Jörg Prante [joergprante@gmail.com](mailto:joergprante@gmail.com) wrote:
> Note that ES thread pools are aligned with the number of CPU cores you
> have. A higher number of threads will not result in higher performance. So,
> if you must really use 500 concurrent threads for best bulk performance,
> you have either the world largest CPU core machine, or your code have some
> other fundamental problems. You should check if you can process 8-10 MB
> bulk data per second per node (for example, monitor the network connection).
>
> The BulkProcessor class is not marked for internal use (as no ES class
> is). I think as long as it is not marked "@deprecated" it will stay in ES
> API (if not I could provide substitution class as a plugin because
> BulkProcessor does no magic, but I think this will not be necessary)
> Remember, the people at ES are really helpful and supportive and high
> performance bulk indexing is a strong requirement for many of us in the
> community.
>
> Jörg
>
> Am 03.07.13 09:14, schrieb Thomas Bolis:
>
> > Dear Jorg, I will try it with less concurrent requests, but the thing is
> > that performance will drop probably I will go to my other implementation
> > which is through BulkRequestBuilder which I don't have these issues.
> >
> > My question is though whether BulkProcessor is a class meant to be used
> > generally or it is an internal class for elasticsearch engine an I probably
> > do not use it well..
>
> --
> You received this message because you are subscribed to a topic in the
> Google Groups "elasticsearch" group.
> To unsubscribe from this topic, visit [https://groups.google.com/d/](https://groups.google.com/d/)\*\*
> topic/elasticsearch/\*\*vUtJKbjb1bo/unsubscribe[https://groups.google.com/d/topic/elasticsearch/vUtJKbjb1bo/unsubscribe](https://groups.google.com/d/topic/elasticsearch/vUtJKbjb1bo/unsubscribe)
> .
> To unsubscribe from this group and all its topics, send an email to
> elasticsearch+unsubscribe@\*\*[googlegroups.com](http://googlegroups.com)[elasticsearch%2Bunsubscribe@googlegroups.com](mailto:elasticsearch%2Bunsubscribe@googlegroups.com)
> .
> For more options, visit [https://groups.google.com/\*\*groups/opt\_out](https://groups.google.com/**groups/opt_out)[https://groups.google.com/groups/opt\_out](https://groups.google.com/groups/opt_out)
> .
--
You received this message because you are subscribed to the Google Groups "elasticsearch" group.
To unsubscribe from this group and stop receiving emails from it, send an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).
For more options, visit [https://groups.google.com/groups/opt\_out](https://groups.google.com/groups/opt_out).
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [July 6, 2017, 2:28am UTC](https://discuss.elastic.co/t/bulkprocessor-usage-is-safe/12625/7 "2017-07-06T02:28:25Z")