# Document ALWAYS Routes to ONE Shard when BULK Loading - HELP! **URL:** \ **Category:** Elasticsearch\ **Created:** [August 23, 2012, 1:15pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827 "2012-08-23T13:15:05Z")\ **Posts on this page:** 19\ **Page:** 1
**Author:** ![amos\_wood](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/amos_wood/32/1880_2.png) [@amos\_wood](https://discuss.elastic.co/u/amos_wood)\ **Post date:** [August 23, 2012, 1:15pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/1 "2012-08-23T13:15:05Z")
We have an index named "partylogins" which we are trying to bulk load using the Java API. The index has 4 shards and no replicas during our bulk load process. However, all documents being indexed are being saved in shard 3 of the index. The \_routing field is not setup on this index and we aren't using any aliases. The code we are using for bulk loading the documents is: ``` private void saveSpecRecords(final String index, final Listrecords ``` ) throws Exception { Client client = null; BulkRequestBuilder bulkRequest = null; long size = 0l; ``` client = getClient(); bulkRequest = client.prepareBulk(); bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); bulkRequest.setRefresh(DEFAULT_REFRESH); bulkRequest.setReplicationType(ReplicationType.ASYNC); bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); bulkRequest.setReplicationType(ReplicationType.ASYNC); bulkRequest.setRefresh(false); for (Record record : records) { Map source_fields = record.getSource(); IndexRequestBuilder doc = client.prepareIndex(index,DEFAULT_DOC_TYPE ``` , record.getId()); doc.setConsistencyLevel(WriteConsistencyLevel.ONE); doc.setRefresh(false); // DEFAULT\_REFRESH [//doc.setCreate](https://doc.setCreate)(true); // DEFAULT\_CREATE doc.setReplicationType(ReplicationType.ASYNC); ``` // Remove the id field from the source and set the source source_fields.remove(XmlReader.ID_FIELD); doc.setSource(source_fields); logger.debug("record id: " + record.getId()); //logger.debug("source keys: " + source_fields.keySet()); // Add the size up to average it out later size += record.getJsonString().length(); bulkRequest.add(doc); } logger.debug("record source size avg (Kbs): " + (size / records.size ``` () / 1024)); ``` ListenableActionFuture lbr = bulkRequest.execute(); logger.debug("start actionGet() for BulkResponse"); BulkResponse response = lbr.actionGet(); logger.debug("end actionGet() for BulkResponse"); if (response.hasFailures()) { logger.error("found the following errors while processing a ``` bulk insert"); for (BulkItemResponse item : response.items()) { if (item.failed()) { logger.debug(item.getIndex()); logger.debug(item.getId()); logger.error(item.failureMessage()); } } throw new Exception("error bulk copying data into spec index " +index ); } } What am I missing? Please help if you have an idea of why all of the documents end up on 1 shard. -- ---
**Author:** ![amos\_wood](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/amos_wood/32/1880_2.png) [@amos\_wood](https://discuss.elastic.co/u/amos_wood)\ **Post date:** [August 23, 2012, 1:31pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/2 "2012-08-23T13:31:15Z")
Also, I have verified that record.getId() has a UNIQUE Id in each value. A sample output of my logging looks like: ... record id: A6855601-7A01-4C20-8F07-AC004BEC415E record id: 89D5E410-BEFB-461B-958E-0138CE03634E record id: EBBBF0C1-B049-46E4-9C04-2B23BCCABE88 record source size avg (Kbs): 17 start actionGet() for BulkResponse end actionGet() for BulkResponse P.S. I hope that this issue is related to the index speed issue which I am seeing too. Using this code and the 17Kb documents, I am only able to INDEX 7 documents/sec which is horribly slow. -- ---
**Author:** ![dadoonet](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/dadoonet/32/137187_2.png) [@dadoonet](https://discuss.elastic.co/u/dadoonet)\ **Post date:** [August 23, 2012, 1:37pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/3 "2012-08-23T13:37:03Z")
I suggest that you execute the bulk request every 100 documents instead of after 17k. That's how rivers are implemented. Then, try to increase bulk size to 200, 300... And so on. HTH David -- David 馃槈 Twitter : @dadoonet / @elasticsearchfr / @scrutmydocs Le 23 ao没t 2012 脿 15:15, "amos.wood" [amos.wood@lifeway.com](mailto:amos.wood@lifeway.com) a 茅crit : > We have an index named "partylogins" which we are trying to bulk load using the Java API. The index has 4 shards and no replicas during our bulk load process. However, all documents being indexed are being saved in shard 3 of the index. The \_routing field is not setup on this index and we aren't using any aliases. > > The code we are using for bulk loading the documents is: > > ``` > private void saveSpecRecords(final String index, final List records) throws Exception { > Client client = null; > BulkRequestBuilder bulkRequest = null; > long size = 0l; > > client = getClient(); > > bulkRequest = client.prepareBulk(); > bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); > bulkRequest.setRefresh(DEFAULT_REFRESH); > bulkRequest.setReplicationType(ReplicationType.ASYNC); > > bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); > bulkRequest.setReplicationType(ReplicationType.ASYNC); > bulkRequest.setRefresh(false); > > for (Record record : records) { > Map source_fields = record.getSource(); > IndexRequestBuilder doc = client.prepareIndex(index, DEFAULT_DOC_TYPE, record.getId()); > doc.setConsistencyLevel(WriteConsistencyLevel.ONE); > doc.setRefresh(false); // DEFAULT_REFRESH > //doc.setCreate(true); // DEFAULT_CREATE > doc.setReplicationType(ReplicationType.ASYNC); > > // Remove the id field from the source and set the source > source_fields.remove(XmlReader.ID_FIELD); > doc.setSource(source_fields); > > logger.debug("record id: " + record.getId()); > //logger.debug("source keys: " + source_fields.keySet()); > > // Add the size up to average it out later > size += record.getJsonString().length(); > > bulkRequest.add(doc); > } > > logger.debug("record source size avg (Kbs): " + (size / records.size() / 1024)); > > ListenableActionFuture lbr = bulkRequest.execute(); > logger.debug("start actionGet() for BulkResponse"); > BulkResponse response = lbr.actionGet(); > logger.debug("end actionGet() for BulkResponse"); > if (response.hasFailures()) { > logger.error("found the following errors while processing a bulk insert"); > for (BulkItemResponse item : response.items()) { > if (item.failed()) { > logger.debug(item.getIndex()); > logger.debug(item.getId()); > logger.error(item.failureMessage()); > } > } > throw new Exception("error bulk copying data into spec index " + index); > } > } > > ``` > > What am I missing? Please help if you have an idea of why all of the documents end up on 1 shard. > > -- -- ---
**Author:** ![Clinton\_Gormley](https://avatars.discourse-cdn.com/v4/letter/c/50afbb/32.png) [@Clinton\_Gormley](https://discuss.elastic.co/u/Clinton_Gormley)\ **Post date:** [August 23, 2012, 1:40pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/4 "2012-08-23T13:40:26Z")
On Thu, 2012-08-23 at 06:15 -0700, amos.wood wrote: > We have an index named "partylogins" which we are trying to bulk load > using the Java API. The index has 4 shards and no replicas during our > bulk load process. However, all documents being indexed are being > saved in shard 3 of the index. The \_routing field is not setup on > this index and we aren't using any aliases. Do your docs use the parent field? Because that implies routing Also, why set indexing to async - you can overwhelm ES if you try to index more than your nodes to handle. If you leave it as sync, then you'll get automatic rate limiting. You can run multiple processes in parallel, but ES will limit the overall rate to what it can handle. Also, David's comment about indexing every 100 - 1000 docs instead of after 17,500 is a good one. Indexing that number of docs can use up lots of memory and slow things down clint > The code we are using for bulk loading the documents is: > > ``` > private void saveSpecRecords(final String index, final > > ``` > > List records) throws Exception { > Client client = null; > BulkRequestBuilder bulkRequest = null; > long size = 0l; > > ``` > client = getClient(); > > bulkRequest = client.prepareBulk(); > bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); > bulkRequest.setRefresh(DEFAULT_REFRESH); > bulkRequest.setReplicationType(ReplicationType.ASYNC); > > bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); > bulkRequest.setReplicationType(ReplicationType.ASYNC); > bulkRequest.setRefresh(false); > > for (Record record : records) { > Map source_fields = record.getSource(); > IndexRequestBuilder doc = client.prepareIndex(index, > > ``` > > DEFAULT\_DOC\_TYPE, record.getId()); > doc.setConsistencyLevel(WriteConsistencyLevel.ONE); > doc.setRefresh(false); // DEFAULT\_REFRESH > [//doc.setCreate](https://doc.setCreate)(true); // DEFAULT\_CREATE > doc.setReplicationType(ReplicationType.ASYNC); > > ``` > // Remove the id field from the source and set the source > source_fields.remove(XmlReader.ID_FIELD); > doc.setSource(source_fields); > > logger.debug("record id: " + record.getId()); > //logger.debug("source keys: " + source_fields.keySet()); > > // Add the size up to average it out later > size += record.getJsonString().length(); > > bulkRequest.add(doc); > } > > logger.debug("record source size avg (Kbs): " + (size / > > ``` > > records.size() / 1024)); > > ``` > ListenableActionFuture lbr = > > ``` > > bulkRequest.execute(); > logger.debug("start actionGet() for BulkResponse"); > BulkResponse response = lbr.actionGet(); > logger.debug("end actionGet() for BulkResponse"); > if (response.hasFailures()) { > logger.error("found the following errors while processing > a bulk insert"); > for (BulkItemResponse item : response.items()) { > if (item.failed()) { > logger.debug(item.getIndex()); > logger.debug(item.getId()); > logger.error(item.failureMessage()); > } > } > throw new Exception("error bulk copying data into spec > index " + index); > } > } > > What am I missing? Please help if you have an idea of why all of the > documents end up on 1 shard. > > -- -- ---
**Author:** ![amos\_wood](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/amos_wood/32/1880_2.png) [@amos\_wood](https://discuss.elastic.co/u/amos_wood)\ **Post date:** [August 23, 2012, 1:41pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/5 "2012-08-23T13:41:06Z")
> I suggest that you execute the bulk request every 100 documents instead of > after 17k. The 17Kb number is the size of the document and not the count. I have tried executing the bulk request with 100 and 1000 and everything between with no success. -- ---
**Author:** ![amos\_wood](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/amos_wood/32/1880_2.png) [@amos\_wood](https://discuss.elastic.co/u/amos_wood)\ **Post date:** [August 23, 2012, 1:45pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/6 "2012-08-23T13:45:58Z")
On Thursday, August 23, 2012 8:40:26 AM UTC-5, Clinton Gormley wrote: > Do your docs use the parent field? > Because that implies routing No. -- ---
**Author:** ![amos\_wood](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/amos_wood/32/1880_2.png) [@amos\_wood](https://discuss.elastic.co/u/amos_wood)\ **Post date:** [August 23, 2012, 1:53pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/7 "2012-08-23T13:53:29Z")
Do you think that this issue has to do with routing? If so, does the \_routing field have to be "enabled" to use the document id as the default? By default, doesn't the routing of the document use a hash of document id field? If so, does it use the "\_id" value from the IndexRequestBuilder:setId() function for routing to the appropriate shard in the index Another oddity is that when not bulk loading the documents and instead just INDEXing every document, all of them still went to shard 3 of the index which implies that it is not a bulk loading issue, right? On Thursday, August 23, 2012 8:15:05 AM UTC-5, amos.wood wrote: > We have an index named "partylogins" which we are trying to bulk load > using the Java API. The index has 4 shards and no replicas during our bulk > load process. However, all documents being indexed are being saved in > shard 3 of the index. The \_routing field is not setup on this index and we > aren't using any aliases. > > The code we are using for bulk loading the documents is: > > ``` > private void saveSpecRecords(final String index, final Listrecords > > ``` > > ) throws Exception { > Client client = null; > BulkRequestBuilder bulkRequest = null; > long size = 0l; > > ``` > client = getClient(); > > bulkRequest = client.prepareBulk(); > bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); > bulkRequest.setRefresh(DEFAULT_REFRESH); > bulkRequest.setReplicationType(ReplicationType.ASYNC); > > bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); > bulkRequest.setReplicationType(ReplicationType.ASYNC); > bulkRequest.setRefresh(false); > > for (Record record : records) { > Map source_fields = record.getSource(); > IndexRequestBuilder doc = client.prepareIndex(index,DEFAULT_DOC_TYPE > > ``` > > , record.getId()); > doc.setConsistencyLevel(WriteConsistencyLevel.ONE); > doc.setRefresh(false); // DEFAULT\_REFRESH > [//doc.setCreate](https://doc.setCreate)(true); // DEFAULT\_CREATE > doc.setReplicationType(ReplicationType.ASYNC); > > ``` > // Remove the id field from the source and set the source > source_fields.remove(XmlReader.ID_FIELD); > doc.setSource(source_fields); > > logger.debug("record id: " + record.getId()); > //logger.debug("source keys: " + source_fields.keySet()); > > // Add the size up to average it out later > size += record.getJsonString().length(); > > bulkRequest.add(doc); > } > > logger.debug("record source size avg (Kbs): " + (size / records. > > ``` > > size() / 1024)); > > ``` > ListenableActionFuture lbr = bulkRequest.execute(); > logger.debug("start actionGet() for BulkResponse"); > BulkResponse response = lbr.actionGet(); > logger.debug("end actionGet() for BulkResponse"); > if (response.hasFailures()) { > logger.error("found the following errors while processing a > > ``` > > bulk insert"); > for (BulkItemResponse item : response.items()) { > if (item.failed()) { > logger.debug(item.getIndex()); > logger.debug(item.getId()); > logger.error(item.failureMessage()); > } > } > throw new Exception("error bulk copying data into spec index " > > - index); > } > } > > What am I missing? Please help if you have an idea of why all of the > documents end up on 1 shard. -- ---
**Author:** ![Clinton\_Gormley](https://avatars.discourse-cdn.com/v4/letter/c/50afbb/32.png) [@Clinton\_Gormley](https://discuss.elastic.co/u/Clinton_Gormley)\ **Post date:** [August 23, 2012, 2:01pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/8 "2012-08-23T14:01:48Z")
On Thu, 2012-08-23 at 06:53 -0700, amos.wood wrote: > Do you think that this issue has to do with routing? Looks like it > If so, does the \_routing field have to be "enabled" to use the > document id as the default? No > By default, doesn't the routing of the document use a hash of document > id field? Yes > If so, does it use the "\_id" value from the > IndexRequestBuilder:setId() function for routing to the appropriate > shard in the index A hash thereof, yes So one possibility is that all of your \_ids are hashing to the same value (which seems unlikely) I'd suggest trying to break the problem down into a smaller version (especially as you know that there is a problem with what you are doing with single doc indexing, rather than bulk) Gist a curl recreation of the entire problem (from index creation onwards), then the issue will quickly become apparent. (it may also become apparent to you before you finish the gist) clint > Another oddity is that when not bulk loading the documents and instead > just INDEXing every document, all of them still went to shard 3 of the > index which implies that it is not a bulk loading issue, right? > > On Thursday, August 23, 2012 8:15:05 AM UTC-5, amos.wood wrote: > We have an index named "partylogins" which we are trying to > bulk load using the Java API. The index has 4 shards and no > replicas during our bulk load process. However, all documents > being indexed are being saved in shard 3 of the index. The > \_routing field is not setup on this index and we aren't using > any aliases. > > ``` > The code we are using for bulk loading the documents is: > > private void saveSpecRecords(final String index, final > List records) throws Exception { > Client client = null; > BulkRequestBuilder bulkRequest = null; > long size = 0l; > > client = getClient(); > > bulkRequest = client.prepareBulk(); > > bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); > bulkRequest.setRefresh(DEFAULT_REFRESH); > bulkRequest.setReplicationType(ReplicationType.ASYNC); > > > bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); > bulkRequest.setReplicationType(ReplicationType.ASYNC); > bulkRequest.setRefresh(false); > > for (Record record : records) { > Map source_fields = > record.getSource(); > IndexRequestBuilder doc = > client.prepareIndex(index, DEFAULT_DOC_TYPE, record.getId()); > > doc.setConsistencyLevel(WriteConsistencyLevel.ONE); > doc.setRefresh(false); // DEFAULT_REFRESH > //doc.setCreate(true); // DEFAULT_CREATE > doc.setReplicationType(ReplicationType.ASYNC); > > // Remove the id field from the source and set the > source > source_fields.remove(XmlReader.ID_FIELD); > doc.setSource(source_fields); > > logger.debug("record id: " + record.getId()); > //logger.debug("source keys: " + > source_fields.keySet()); > > // Add the size up to average it out later > size += record.getJsonString().length(); > > > bulkRequest.add(doc); > } > > logger.debug("record source size avg (Kbs): " + > (size / records.size() / 1024)); > > ListenableActionFuture lbr = > bulkRequest.execute(); > logger.debug("start actionGet() for BulkResponse"); > BulkResponse response = lbr.actionGet(); > logger.debug("end actionGet() for BulkResponse"); > if (response.hasFailures()) { > logger.error("found the following errors while > processing a bulk insert"); > for (BulkItemResponse item : response.items()) { > if (item.failed()) { > logger.debug(item.getIndex()); > logger.debug(item.getId()); > logger.error(item.failureMessage()); > } > } > throw new Exception("error bulk copying data into > spec index " + index); > } > } > > > > > What am I missing? Please help if you have an idea of why all > of the documents end up on 1 shard. > > ``` > > -- -- ---
**Author:** ![dadoonet](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/dadoonet/32/137187_2.png) [@dadoonet](https://discuss.elastic.co/u/dadoonet)\ **Post date:** [August 23, 2012, 2:20pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/9 "2012-08-23T14:20:43Z")
> it may also become apparent to you before you finish the gist > +1000 I hate that. Often, I started to create a nice Gist to expose my issues with full explanation and comments and the solution came by the question itself! 馃槈 -\> just end with throwing the email in the trashcan ! 馃槈 -- David 馃槈 Twitter : @dadoonet / @elasticsearchfr / @scrutmydocs Le 23 ao没t 2012 脿 16:01, Clinton Gormley [clint@traveljury.com](mailto:clint@traveljury.com) a 茅crit : On Thu, 2012-08-23 at 06:53 -0700, amos.wood wrote: > Do you think that this issue has to do with routing? Looks like it > If so, does the \_routing field have to be "enabled" to use the > document id as the default? No > By default, doesn't the routing of the document use a hash of document > id field? Yes > If so, does it use the "\_id" value from the > IndexRequestBuilder:setId() function for routing to the appropriate > shard in the index A hash thereof, yes So one possibility is that all of your \_ids are hashing to the same value (which seems unlikely) I'd suggest trying to break the problem down into a smaller version (especially as you know that there is a problem with what you are doing with single doc indexing, rather than bulk) Gist a curl recreation of the entire problem (from index creation onwards), then the issue will quickly become apparent. (it may also become apparent to you before you finish the gist) clint > Another oddity is that when not bulk loading the documents and instead > just INDEXing every document, all of them still went to shard 3 of the > index which implies that it is not a bulk loading issue, right? > > On Thursday, August 23, 2012 8:15:05 AM UTC-5, amos.wood wrote: > We have an index named "partylogins" which we are trying to > bulk load using the Java API. The index has 4 shards and no > replicas during our bulk load process. However, all documents > being indexed are being saved in shard 3 of the index. The > \_routing field is not setup on this index and we aren't using > any aliases. > > ``` > The code we are using for bulk loading the documents is: > > private void saveSpecRecords(final String index, final > List records) throws Exception { > Client client = null; > BulkRequestBuilder bulkRequest = null; > long size = 0l; > > client = getClient(); > > bulkRequest = client.prepareBulk(); > > bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); > bulkRequest.setRefresh(DEFAULT_REFRESH); > bulkRequest.setReplicationType(ReplicationType.ASYNC); > > bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); > bulkRequest.setReplicationType(ReplicationType.ASYNC); > bulkRequest.setRefresh(false); > > for (Record record : records) { > Map source_fields = > record.getSource(); > IndexRequestBuilder doc = > client.prepareIndex(index, DEFAULT_DOC_TYPE, record.getId()); > > doc.setConsistencyLevel(WriteConsistencyLevel.ONE); > doc.setRefresh(false); // DEFAULT_REFRESH > //doc.setCreate(true); // DEFAULT_CREATE > doc.setReplicationType(ReplicationType.ASYNC); > > // Remove the id field from the source and set the > source > source_fields.remove(XmlReader.ID_FIELD); > doc.setSource(source_fields); > > logger.debug("record id: " + record.getId()); > //logger.debug("source keys: " + > source_fields.keySet()); > > // Add the size up to average it out later > size += record.getJsonString().length(); > > bulkRequest.add(doc); > } > > logger.debug("record source size avg (Kbs): " + > (size / records.size() / 1024)); > > ListenableActionFuture lbr = > bulkRequest.execute(); > logger.debug("start actionGet() for BulkResponse"); > BulkResponse response = lbr.actionGet(); > logger.debug("end actionGet() for BulkResponse"); > if (response.hasFailures()) { > logger.error("found the following errors while > processing a bulk insert"); > for (BulkItemResponse item : response.items()) { > if (item.failed()) { > logger.debug(item.getIndex()); > logger.debug(item.getId()); > logger.error(item.failureMessage()); > } > } > throw new Exception("error bulk copying data into > spec index " + index); > } > } > > What am I missing? Please help if you have an idea of why all > of the documents end up on 1 shard. > > ``` > > -- -- -- ---
**Author:** ![Wes\_Plunk](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/wes_plunk/32/2446_2.png) [@Wes\_Plunk](https://discuss.elastic.co/u/Wes_Plunk)\ **Post date:** [August 23, 2012, 2:58pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/10 "2012-08-23T14:58:04Z")
That would be nice, we're working on the gist now. Will provide results here either way On Thursday, August 23, 2012 9:20:43 AM UTC-5, David Pilato wrote: > > it may also become apparent to you before you finish the gist > > +1000 > > I hate that. Often, I started to create a nice Gist to expose my issues > with full explanation and comments and the solution came by the question > itself! 馃槈 -\> just end with throwing the email in the trashcan ! 馃槈 > > -- > David 馃槈 > Twitter : @dadoonet / @elasticsearchfr / @scrutmydocs > > Le 23 ao没t 2012 脿 16:01, Clinton Gormley \<[cl...@traveljury.com](mailto:cl...@traveljury.com)\\> > a 茅crit : > > On Thu, 2012-08-23 at 06:53 -0700, amos.wood wrote: > > > Do you think that this issue has to do with routing? > > Looks like it > > > If so, does the \_routing field have to be "enabled" to use the > > document id as the default? > > No > > > By default, doesn't the routing of the document use a hash of document > > id field? > > Yes > > > If so, does it use the "\_id" value from the > > IndexRequestBuilder:setId() function for routing to the appropriate > > shard in the index > > A hash thereof, yes > > So one possibility is that all of your \_ids are hashing to the same > value (which seems unlikely) > > I'd suggest trying to break the problem down into a smaller version > (especially as you know that there is a problem with what you are doing > with single doc indexing, rather than bulk) > > Gist a curl recreation of the entire problem (from index creation > onwards), then the issue will quickly become apparent. (it may also > become apparent to you before you finish the gist) > > clint > > > Another oddity is that when not bulk loading the documents and instead > > just INDEXing every document, all of them still went to shard 3 of the > > index which implies that it is not a bulk loading issue, right? > > > > On Thursday, August 23, 2012 8:15:05 AM UTC-5, amos.wood wrote: > > We have an index named "partylogins" which we are trying to > > bulk load using the Java API. The index has 4 shards and no > > replicas during our bulk load process. However, all documents > > being indexed are being saved in shard 3 of the index. The > > \_routing field is not setup on this index and we aren't using > > any aliases. > > > > ``` > > The code we are using for bulk loading the documents is: > > > > private void saveSpecRecords(final String index, final > > List records) throws Exception { > > Client client = null; > > BulkRequestBuilder bulkRequest = null; > > long size = 0l; > > > > client = getClient(); > > > > bulkRequest = client.prepareBulk(); > > > > bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); > > bulkRequest.setRefresh(DEFAULT_REFRESH); > > bulkRequest.setReplicationType(ReplicationType.ASYNC); > > > > bulkRequest.setConsistencyLevel(WriteConsistencyLevel.ONE); > > bulkRequest.setReplicationType(ReplicationType.ASYNC); > > bulkRequest.setRefresh(false); > > > > for (Record record : records) { > > Map source_fields = > > record.getSource(); > > IndexRequestBuilder doc = > > client.prepareIndex(index, DEFAULT_DOC_TYPE, record.getId()); > > > > doc.setConsistencyLevel(WriteConsistencyLevel.ONE); > > doc.setRefresh(false); // DEFAULT_REFRESH > > //doc.setCreate(true); // DEFAULT_CREATE > > doc.setReplicationType(ReplicationType.ASYNC); > > > > // Remove the id field from the source and set the > > source > > source_fields.remove(XmlReader.ID_FIELD); > > doc.setSource(source_fields); > > > > logger.debug("record id: " + record.getId()); > > //logger.debug("source keys: " + > > source_fields.keySet()); > > > > // Add the size up to average it out later > > size += record.getJsonString().length(); > > > > bulkRequest.add(doc); > > } > > > > logger.debug("record source size avg (Kbs): " + > > (size / records.size() / 1024)); > > > > ListenableActionFuture lbr = > > bulkRequest.execute(); > > logger.debug("start actionGet() for BulkResponse"); > > BulkResponse response = lbr.actionGet(); > > logger.debug("end actionGet() for BulkResponse"); > > if (response.hasFailures()) { > > logger.error("found the following errors while > > processing a bulk insert"); > > for (BulkItemResponse item : response.items()) { > > if (item.failed()) { > > logger.debug(item.getIndex()); > > logger.debug(item.getId()); > > logger.error(item.failureMessage()); > > } > > } > > throw new Exception("error bulk copying data into > > spec index " + index); > > } > > } > > > > What am I missing? Please help if you have an idea of why all > > of the documents end up on 1 shard. > > > > ``` > > > > -- > > -- -- ---
**Author:** ![Wes\_Plunk](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/wes_plunk/32/2446_2.png) [@Wes\_Plunk](https://discuss.elastic.co/u/Wes_Plunk)\ **Post date:** [August 23, 2012, 2:58pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/11 "2012-08-23T14:58:37Z")
That would be nice, we're working on the gist now. Will provide results here either way -- ---
**Author:** ![Wes\_Plunk](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/wes_plunk/32/2446_2.png) [@Wes\_Plunk](https://discuss.elastic.co/u/Wes_Plunk)\ **Post date:** [August 23, 2012, 5:50pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/12 "2012-08-23T17:50:05Z")
Here's the gist: [https://gist.github.com/3439162](https://gist.github.com/3439162) If you need more data please just let us know. Our conclusion from this example is that using numeric ids vs guids is causing the problem. Thoughts as to why? Does this run and perform the same on your system? We have 2 nodes on one cluster and 4 on another and this example produced the same results on both clusters Our definition of failure: It fails when all records are written to the same shard regardless of how the shards are allocated to the nodes -- ---
**Author:** ![dadoonet](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/dadoonet/32/137187_2.png) [@dadoonet](https://discuss.elastic.co/u/dadoonet)\ **Post date:** [August 23, 2012, 6:01pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/13 "2012-08-23T18:01:15Z")
How did you generate guids? The way I understood shard repartition is that everything is based on internal guid computed for each id. Could it be a coincidence that you generated only guids that will be routed to the same shard? -- David 馃槈 Twitter : @dadoonet / @elasticsearchfr / @scrutmydocs Le 23 ao没t 2012 脿 19:50, Wes Plunk [wes@wesandemily.com](mailto:wes@wesandemily.com) a 茅crit : Here's the gist: [https://gist.github.com/3439162](https://gist.github.com/3439162) If you need more data please just let us know. Our conclusion from this example is that using numeric ids vs guids is causing the problem. Thoughts as to why? Does this run and perform the same on your system? We have 2 nodes on one cluster and 4 on another and this example produced the same results on both clusters ## Our definition of failure: It fails when all records are written to the same shard regardless of how the shards are allocated to the nodes -- ---
**Author:** ![Clinton\_Gormley](https://avatars.discourse-cdn.com/v4/letter/c/50afbb/32.png) [@Clinton\_Gormley](https://discuss.elastic.co/u/Clinton_Gormley)\ **Post date:** [August 23, 2012, 6:09pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/14 "2012-08-23T18:09:07Z")
Hi Wes On Thu, 2012-08-23 at 10:50 -0700, Wes Plunk wrote: > Here's the gist: [unbalanced docs in shards 路 GitHub](https://gist.github.com/3439162) > > If you need more data please just let us know. Our conclusion from > this example is that using numeric ids vs guids is causing the > problem. Thoughts as to why? Does this run and perform the same on > your system? This does exactly as you describe on our system. I also tried changing a few IDs (adding a character) and then it spread them onto more nodes. So bizarrely, it seems that almost all your IDs are being hashed to the same shard, which is either an amazing coincidence or a bug, I'd say. Please open an issue linking to this gist ta clint > We have 2 nodes on one cluster and 4 on another and this example > produced the same results on both clusters > > Our definition of failure: It fails when all records are written to > the same shard regardless of how the shards are allocated to the nodes > > -- -- ---
**Author:** ![drewr](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/drewr/32/7803_2.png) [@drewr](https://discuss.elastic.co/u/drewr)\ **Post date:** [August 23, 2012, 9:54pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/15 "2012-08-23T21:54:11Z")
Clinton Gormley wrote: > On Thu, 2012-08-23 at 10:50 -0700, Wes Plunk wrote: > > Here's the gist: [unbalanced docs in shards 路 GitHub](https://gist.github.com/3439162) > > > > If you need more data please just let us know. Our conclusion from > > this example is that using numeric ids vs guids is causing the > > problem. Thoughts as to why? Does this run and perform the same on > > your system? > > This does exactly as you describe on our system. I also tried > changing a few IDs (adding a character) and then it spread them > onto more nodes. > > So bizarrely, it seems that almost all your IDs are being hashed to > the same shard, which is either an amazing coincidence or a bug, > I'd say. Interesting! The hash calculation (based on DjbHashFunction) does produce different ints for each uuid: [http://p.draines.com/13457538251189dc8c3cc.txt](http://p.draines.com/13457538251189dc8c3cc.txt) However, for this case, it happens to produce ints that are always divisible by 4. This causes the modulo calculation in PlainOperationRouting.shardId() to always return the same number (in my case it's 0). Math.abs(h % n) I've played around with different scenarios and it's kind of hit-or-miss. Different 36-byte ids with n that's a multiple of 2 doesn't get a good random distribution. For example, here's n == 32 (taking only 40 ids): (4 4 0 20 4 12 20 28 8 0 8 0 28 12 20 0 0 28 12 24 28 12 0 4 28 28 4 28 12 24 28 12 4 16 4 20 12 24 8 24) But n == 33 is: (21 21 21 1 11 1 11 15 29 25 2 2 30 21 14 10 12 10 31 12 8 20 19 4 18 21 18 7 7 10 28 14 32 27 13 28 10 31 11 24) ...which is a little better. I'm no crypto expert but it may be that this is a "limitation" of the djb hash function. I'm not sure what the right fix is for ES. A provision might need to be put in place for this exact scenario where it appends some random bytes to the end of the id to avoid this particular bug. From a practical standpoint, Wes & Amos, you guys can Base64 your id into a 48-byte string like this which will distribute more evenly (using Base64.encodeBytes(id.getBytes())): (1 2 3 2 2 0 3 3 1 1 0 3 0 2 0 3 3 3 3 3 3 0 0 3 2 3 0 3 3 2 2 0 1 1 0 1 2 2 1 1) If your UUIDs aren't meaningful apart from this index, then you can use an ES-generated 22-byte UUID.randomBase64UUID to save disk and ram. 40 of these with 4 shards yields this distribution: (3 0 0 3 0 1 3 1 2 1 0 3 1 2 2 2 0 1 1 0 2 2 3 1 3 0 2 3 2 0 0 0 1 3 3 3 2 0 1 0) -Drew -- ---
**Author:** ![amos\_wood](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/amos_wood/32/1880_2.png) [@amos\_wood](https://discuss.elastic.co/u/amos_wood)\ **Post date:** [August 24, 2012, 12:19pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/16 "2012-08-24T12:19:52Z")
Drew -- Thanks for the info about the djb hash function. In our case, the ID is meaningful because it is how that get/update the document and how that the documents are joined together. Would it be possible to solve our case by changing the "n" in the equation from 4 shards to something like 5 or 7? Would that tend to balance things better? -- ---
**Author:** ![drewr](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/drewr/32/7803_2.png) [@drewr](https://discuss.elastic.co/u/drewr)\ **Post date:** [August 24, 2012, 12:58pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/17 "2012-08-24T12:58:05Z")
amos.wood wrote: > Drew -- Thanks for the info about the djb hash function. > > In our case, the ID is meaningful because it is how that get/update > the document and how that the documents are joined together. > > Would it be possible to solve our case by changing the "n" in the > equation from 4 shards to something like 5 or 7? Would that tend to > balance things better? Yes! Forgot to mention that. Changing the shard count will also better distribute documents. An odd number, like you say 5 or 7, would work best. -Drew -- ---
**Author:** ![Ivan](https://avatars.discourse-cdn.com/v4/letter/i/df788c/32.png) [@Ivan](https://discuss.elastic.co/u/Ivan)\ **Post date:** [August 24, 2012, 8:34pm UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/18 "2012-08-24T20:34:45Z")
Very interesting read! Pluggable hash functions would be interesting (and dangerous at the same time). -- Ivan On Thu, Aug 23, 2012 at 2:54 PM, Drew Raines [aaraines@gmail.com](mailto:aaraines@gmail.com) wrote: > Clinton Gormley wrote: > > > On Thu, 2012-08-23 at 10:50 -0700, Wes Plunk wrote: > > > > Here's the gist: [unbalanced docs in shards 路 GitHub](https://gist.github.com/3439162) > > > > > > If you need more data please just let us know. Our conclusion from > > > this example is that using numeric ids vs guids is causing the > > > problem. Thoughts as to why? Does this run and perform the same on > > > your system? > > > > This does exactly as you describe on our system. I also tried > > changing a few IDs (adding a character) and then it spread them > > onto more nodes. > > > > So bizarrely, it seems that almost all your IDs are being hashed to > > the same shard, which is either an amazing coincidence or a bug, > > I'd say. > > Interesting! > > The hash calculation (based on DjbHashFunction) does produce > different ints for each uuid: > > [http://p.draines.com/13457538251189dc8c3cc.txt](http://p.draines.com/13457538251189dc8c3cc.txt) > > However, for this case, it happens to produce ints that are always > divisible by 4. This causes the modulo calculation in > PlainOperationRouting.shardId() to always return the same number (in > my case it's 0). > > Math.abs(h % n) > > I've played around with different scenarios and it's kind of > hit-or-miss. Different 36-byte ids with n that's a multiple of 2 > doesn't get a good random distribution. For example, here's n == 32 > (taking only 40 ids): > > (4 4 0 20 4 12 20 28 8 0 8 0 28 12 20 0 0 28 12 24 28 12 0 4 28 28 4 28 12 24 28 12 4 16 4 20 12 24 8 24) > > But n == 33 is: > > (21 21 21 1 11 1 11 15 29 25 2 2 30 21 14 10 12 10 31 12 8 20 19 4 18 21 18 7 7 10 28 14 32 27 13 28 10 31 11 24) > > ...which is a little better. > > I'm no crypto expert but it may be that this is a "limitation" of the > djb hash function. I'm not sure what the right fix is for ES. A > provision might need to be put in place for this exact scenario where > it appends some random bytes to the end of the id to avoid this > particular bug. > > From a practical standpoint, Wes & Amos, you guys can Base64 your id > into a 48-byte string like this which will distribute more evenly > (using Base64.encodeBytes(id.getBytes())): > > (1 2 3 2 2 0 3 3 1 1 0 3 0 2 0 3 3 3 3 3 3 0 0 3 2 3 0 3 3 2 2 0 1 1 0 1 2 2 1 1) > > If your UUIDs aren't meaningful apart from this index, then you can > use an ES-generated 22-byte UUID.randomBase64UUID to save disk and > ram. 40 of these with 4 shards yields this distribution: > > (3 0 0 3 0 1 3 1 2 1 0 3 1 2 2 2 0 1 1 0 2 2 3 1 3 0 2 3 2 0 0 0 1 3 3 3 2 0 1 0) > > -Drew > > -- -- ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [July 6, 2017, 3:15am UTC](https://discuss.elastic.co/t/document-always-routes-to-one-shard-when-bulk-loading-help/8827/19 "2017-07-06T03:15:07Z")