Elasticsearch-py: bulk index connection timeout caused by readtimeout error

[fantastao]

I'm trying to index using bulk index api in es.But index processing seems to get slower and slower even cause connection timeout.The next is my python code.

def bulk_index():
    chunk = []
    for count, doc in enumerate(doc_generator()):
        id = doc.get('_id')
        chunk.append({
            'index': {
                '_id': id
            }
        })
        chunk.append(doc)
        if (count + 1) % 10000 == 0:
            assert len(chunk) == 20000
            res = es.bulk(index='weibo', doc_type='user', body=chunk)
            assert res['errors'] is False
            print 'keke'
            chunk = []
    print 'docs total count: %s' % (count + 1)

[HonzaKral]

Bulk indexing can be an expensive process so it can indeed take a long time. If it takes more than 10 seconds the only soution is to raise the timeout parameter of the es client. Either for the whole instance or you can specify request_timeout=30 as part of the es.bulk call. There is no way to speed it up from the client - either send smaller bulk requests or raise the timeout value.

also note that you are replicating the chunking logic that we already have as part of the bulk helper: http://elasticsearch-py.readthedocs.org/en/latest/helpers.html#elasticsearch.helpers.bulk

Hope this helps...

[fantastao]

Thanks a lot! Actually what i just wonder,in my case, is that why the first 10000 docs index process is fast, but as time goes by,it gets slower eventually cause timeout.Now that bulk index(es.bulk) has no high performance,why should we use it instead of simple index which I mean es.index?

[HonzaKral]

That is a question you have to ask your elasticsearch cluster, it might be that it's busy, there are many bulk requests that cause the slowdown etc, causes may be many. Monitor your cluster to see what's going on during the bulk indexing to see what's causing it in your case.

bulk is MUCH faster than individual index calls, but it's still kind of expensive compared to get or other operations.

Btw your code will also cause the last chunk (that might be incomplete) to be ignored unless you have exactly N*10000 documents in your dataset.

[HonzaKral]

Closing this since the issue is resolved.

[arianamiri]

Monitor your cluster to see what's going on during the bulk indexing to see what's causing it in your case.

@HonzaKral, any suggestions on what to monitor or what APIs to use to monitor what is going on during a bulk index?

[Ads7]

@HonzaKral comment upon adding timeout to the base client
Elasticsearch(hosts=["127.0.0.1:9200"], timeout=5000)

but received the error
POST http://localhost:8080/_bulk?timeout=30 [status:400 request:0.001s] Traceback (most recent call last): File "workspace/process_data.py", line 86, in <module> indexer.start(PATH + file_name, key, key, date) File "workspace/process_data.py", line 57, in start self.index_df(df,index_name) File "workspace/process_data.py", line 44, in index_df elasticsearch.helpers.bulk(self.es, actions,timeout=30) File "/csnzoo/amansingh/.envs/dev_egm/lib/python3.6/site-packages/elasticsearch/helpers/__init__.py", line 257, in bulk for ok, item in streaming_bulk(client, actions, **kwargs): File "/csnzoo/amansingh/.envs/dev_egm/lib/python3.6/site-packages/elasticsearch/helpers/__init__.py", line 192, in streaming_bulk raise_on_error, **kwargs) File "/csnzoo/amansingh/.envs/dev_egm/lib/python3.6/site-packages/elasticsearch/helpers/__init__.py", line 99, in _process_bulk_chunk raise e File "/csnzoo/amansingh/.envs/dev_egm/lib/python3.6/site-packages/elasticsearch/helpers/__init__.py", line 95, in _process_bulk_chunk resp = client.bulk('\n'.join(bulk_actions) + '\n', **kwargs) File "/csnzoo/amansingh/.envs/dev_egm/lib/python3.6/site-packages/elasticsearch/client/utils.py", line 76, in _wrapped return func(*args, params=params, **kwargs) File "/csnzoo/amansingh/.envs/dev_egm/lib/python3.6/site-packages/elasticsearch/client/__init__.py", line 1150, in bulk headers={'content-type': 'application/x-ndjson'}) File "/csnzoo/amansingh/.envs/dev_egm/lib/python3.6/site-packages/elasticsearch/transport.py", line 314, in perform_request status, headers_response, data = connection.perform_request(method, url, params, body, headers=headers, ignore=ignore, timeout=timeout) File "/csnzoo/amansingh/.envs/dev_egm/lib/python3.6/site-packages/elasticsearch/connection/http_urllib3.py", line 180, in perform_request self._raise_error(response.status, raw_data) File "/csnzoo/amansingh/.envs/dev_egm/lib/python3.6/site-packages/elasticsearch/connection/base.py", line 125, in _raise_error raise HTTP_EXCEPTIONS.get(status_code, TransportError)(status_code, error_message, additional_info) elasticsearch.exceptions.RequestError: TransportError(400, 'parse_exception', 'failed to parse setting [timeout] with value [30] as a time value: unit is missing or unrecognized')

[HonzaKral]

@arianamiri monitor the thread pools (via .cat.thread_pool('bulk')) and the overall health via standard monitoring

@Ads7 The exception yhat you posted has nothing to do with the call to Elasticsearch but instead refers to the line in your code where you specified elasticsearch.helpers.bulk(self.es, actions,timeout=30) which is not supported, you need to remove the timeout kwarg from that call.

[vovavovavovavova]

is this the issue with read timeout?

elasticsearch.helpers.bulk(es,resultArray, request_timeout = (10,10**(-10)) )

elasticsearch.exceptions.ConnectionError: ConnectionError(Timeout value connect
was (10, 1e-10), but it must be an int, float or None.) caused by: ValueError(Ti
meout value connect was (10, 1e-10), but it must be an int, float or None.)

btw, with single request insert like :
resp = requests.put("http://url/my_index/my_doc_type/my_id", data = json.dumps(data), timeout = (10, 10**(-10)))
all works fine. So there is a way to specify "tuple" of just "read_timeout"

[fxdgear]

@vovavovavovavova as of right now we do not support tuple timeouts in elasticsearch-py. Though if that's something you feel we should add you can create an issue and I'll look into it.

Personally I've never used tuples as timeouts. But I do see that there'd be some value there on the requests side. urllib3 does not support this and as I write this I'm not sure how much work it'd be to do to ensure that it works with requests and not with urllib3. But am willing to investigate.