[FEATURE] es.batch.write.wait - time to wait between each bulk

[roeyshemtov]

What kind an issue is this?

• Feature Request.

Feature description

I am running a scenario that i have a big a dataset that i want to index into elastic, but i can't controll the index rate without changes the number of partitions of the dataset(that mean lower/higher parallel writing).
I know there is bulk size and bulk entries but it just adjust the bulking.
what im suggest is to add a configurable es.batch.write.wait, and that is the time in ms that it`ll wait between each bulk.

This configuration should help to adjust too fast index rate without chaning the dataset itself.

What do you think, it will be helpful for other users? because this feature will be great at our usecase.

[jbaiera]

Sorry for the long wait on a response here. I think this is an interesting idea. Many parts of Elasticsearch's internal components have throttling to adjust how much work they're performing to avoid impact on the system, but the throttling that Elasticsearch implements is to avoid internal contention for resources on the Elasticsearch nodes themselves. Another place I can find throttling logic is in Logstash, though it seems that feature is more of a traffic control filter meant to cap the throughput of data that you expect to be within a certain number of events per time unit.

My main concern here is that while introducing throttling options would be a straightforward enhancement, is it the correct action to take? Throttling a job would indicate (to me) that Elasticsearch is not effectively handling the ingestion load, which I feel is a more pressing concern to address in terms of scalability.

Obviously, everyone's use case is unique, so I'm interested to hear what kind of issues you've run into, and how adding a throttle here would assist in fixing those issues vs. improving the general throughput of the system.

[roeyshemtov]

Hello @jbaiera and thanks for answer.
I am partly agree that the infrastructure need to handle it, but in our case we cant improve the general throughput of the system, and we dont care that our Spark ETLs will take longer.
We are two teams that using elasticsearch 5 and we facing two problems that this issue can fix:

1. My problem - we have an infrastructure team (elasticsearch environment admins), that doesn`t allow us to pass a MAX limit of index rate of the environment - so here it can help.
2. Copartenr problems - They running a little cluster of elasticsearch by their own, and when they write their data (~2B rows) the elasticsearch cluster is just crash. (They doesnt care the ETL will take longer and they cant increase their environment - they have enough storage for the data).

[masseyke]

Hi @roeyshemtov. I've usually seen this handled by changing the number of partitions in your spark job (which you mention) or the number of reducers in a mapreduce job. Is there a reason that that is not an option for you? I'm just trying to get a sense for whether this new configuration is a "nice to have" or if it is something that is truly required. Thanks.

[bitdiscovery]

I am also seeing a need for this feature. In my case, the Spark worker nodes have limited capacity and using a small number of partitions in order to throttle down the speed of writes to ES causes the workers to run out of disk. As a result I have nearly 100% CPU on all my ES nodes during ES data-loading. A simple wait configuration between bulk writes would solve it!

[masseyke]

Hi @bitdiscovery is this because the partitions are so large? Or am I misunderstanding? I would have thought that would have caused memory problems before disk problems, and could be solved by repartitioning (which is admittedly not free).

[bitdiscovery]

Thanks for the quick response @masseyke. I believe it is a memory problem but it is just manifesting as an out of disk issue due to Spill() operations (Spark shuffling).

So it is interesting, because I am in this place where if I add more partitions it solves the Spark issue but then I also wind up with more concurrent ES bulk writers, which winds up slamming my cluster even harder. So I am just left with using larger workers with more RAM/Disk which is pricey.

What I am really trying to do is to have N partitions on the Spark side, and N shards on the ES side... I'd like to then provision the ES cluster so that I have some Idle CPU if I do so. However this has been elusive to achieve because too few shards makes my Spark job slow down, but too many makes ES buckle under the write load.

[masseyke]

Yeah unfortunately I think you have to do this as two spark jobs currently, which is not great. The reason is that (if I'm remembering correctly) spark will not let you change the number of executors mid job. So you have one spark job that does whatever processing and enrichment you're doing currently, using as many executors as possible, that writes to HDFS at the end. Then you have a second job that picks those files up with a lot of partitions but only a small number of executors (one per shard you're loading into is probably not a bad starting point, but you can probably go up a little from there).