Hi I am trying to insert json files to my elasticsearch server. I am using python client 7.9.1.
My data is separated into files which are consisted from 1000 documents. I also tried with 500 documents. I am traversing in the folder and inserting files with elasticsearch.helpers.bulk function. At some point it works but on 5th iteration it throws connection timeout. I am also giving 3 data nodes as a list to NODES. I also tried with different timeout values and setting refresh_interval to -1 on elastic server.
class ElasticServiceAlibaba:
def __init__(self) -> None:
self.elastic_client = Elasticsearch(NODES, timeout=60, max_retries=3, retry_on_timeout=True)
def read_from_file(self, filename):
with open(f"{RESULT_PATH}/{filename}", "r") as file:
cases = json.loads(file.read())
return cases
def bulk_insert_to_alibaba(self, cases):
def generate_data():
for case in cases:
yield {
"_index": INDEX_NAME,
"_source": case
}
elasticsearch.helpers.bulk(self.elastic_client, generate_data())
def get_files():
result = [f for f in listdir(RESULT_PATH) if isfile(join(RESULT_PATH, f))]
print(f"Files were reading... File count: {len(result)}")
return result
def write_to_alibaba():
# Write to alibaba
elastic_alibaba = ElasticServiceAlibaba()
files = get_files()
for filename in files:
print(f"Starting to upload {filename}")
cases = elastic_alibaba.read_from_file(filename)
elastic_alibaba.bulk_insert_to_alibaba(cases)
print(f"{filename} completed.")
if __name__ == '__main__':
write_to_alibaba()
print("Completed.")
Output is as following:
Files were reading... File count: 890
Starting to upload result_352.json
result_352.json completed.
Starting to upload result_702.json
result_702.json completed.
Starting to upload result_217.json
result_217.json completed.
Starting to upload result_647.json
result_647.json completed.
Starting to upload result_581.json
Traceback (most recent call last):
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/urllib3/connectionpool.py", line 703, in urlopen
httplib_response = self._make_request(
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/urllib3/connectionpool.py", line 398, in _make_request
conn.request(method, url, **httplib_request_kw)
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/urllib3/connection.py", line 239, in request
super(HTTPConnection, self).request(method, url, body=body, headers=headers)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/http/client.py", line 1252, in request
self._send_request(method, url, body, headers, encode_chunked)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/http/client.py", line 1298, in _send_request
self.endheaders(body, encode_chunked=encode_chunked)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/http/client.py", line 1247, in endheaders
self._send_output(message_body, encode_chunked=encode_chunked)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/http/client.py", line 1046, in _send_output
self.send(chunk)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/http/client.py", line 968, in send
self.sock.sendall(data)
socket.timeout: timed out
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/elasticsearch/connection/http_urllib3.py", line 245, in perform_request
response = self.pool.urlopen(
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/urllib3/connectionpool.py", line 787, in urlopen
retries = retries.increment(
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/urllib3/util/retry.py", line 525, in increment
raise six.reraise(type(error), error, _stacktrace)
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/urllib3/packages/six.py", line 769, in reraise
raise value.with_traceback(tb)
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/urllib3/connectionpool.py", line 703, in urlopen
httplib_response = self._make_request(
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/urllib3/connectionpool.py", line 398, in _make_request
conn.request(method, url, **httplib_request_kw)
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/urllib3/connection.py", line 239, in request
super(HTTPConnection, self).request(method, url, body=body, headers=headers)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/http/client.py", line 1252, in request
self._send_request(method, url, body, headers, encode_chunked)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/http/client.py", line 1298, in _send_request
self.endheaders(body, encode_chunked=encode_chunked)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/http/client.py", line 1247, in endheaders
self._send_output(message_body, encode_chunked=encode_chunked)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/http/client.py", line 1046, in _send_output
self.send(chunk)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/http/client.py", line 968, in send
self.sock.sendall(data)
urllib3.exceptions.ProtocolError: ('Connection aborted.', timeout('timed out'))
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/main.py", line 136, in <module>
write_to_alibaba()
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/main.py", line 129, in write_to_alibaba
elastic_alibaba.bulk_insert_to_alibaba(cases)
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/main.py", line 104, in bulk_insert_to_alibaba
elasticsearch.helpers.bulk(self.elastic_client, generate_data())
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/elasticsearch/helpers/actions.py", line 373, in bulk
for ok, item in streaming_bulk(client, actions, *args, **kwargs):
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/elasticsearch/helpers/actions.py", line 294, in streaming_bulk
for data, (ok, info) in zip(
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/elasticsearch/helpers/actions.py", line 230, in _process_bulk_chunk
for item in gen:
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/elasticsearch/helpers/actions.py", line 179, in _process_bulk_chunk_error
raise error
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/elasticsearch/helpers/actions.py", line 218, in _process_bulk_chunk
resp = client.bulk("\n".join(bulk_actions) + "\n", *args, **kwargs)
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/elasticsearch/client/utils.py", line 152, in _wrapped
return func(*args, params=params, headers=headers, **kwargs)
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/elasticsearch/client/__init__.py", line 444, in bulk
return self.transport.perform_request(
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/elasticsearch/transport.py", line 390, in perform_request
raise e
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/elasticsearch/transport.py", line 358, in perform_request
status, headers_response, data = connection.perform_request(
File "/Users/erdem.ontas/PycharmProjects/pythonProject3/venv/lib/python3.8/site-packages/elasticsearch/connection/http_urllib3.py", line 258, in perform_request
raise ConnectionError("N/A", str(e), e)
elasticsearch.exceptions.ConnectionError: ConnectionError(('Connection aborted.', timeout('timed out'))) caused by: ProtocolError(('Connection aborted.', timeout('timed out')))