From accb19ab905cc68bcc1690e19f3cd3ad18e7f670 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Kr=C3=A1l?= Date: Tue, 18 Jul 2017 15:45:43 -0400 Subject: [PATCH] Action must be a dict, not serialized string --- elasticsearch/helpers/__init__.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/elasticsearch/helpers/__init__.py b/elasticsearch/helpers/__init__.py index cda89262..8bce65f6 100644 --- a/elasticsearch/helpers/__init__.py +++ b/elasticsearch/helpers/__init__.py @@ -55,6 +55,7 @@ def _chunk_actions(actions, chunk_size, max_chunk_bytes, serializer): bulk_actions, bulk_data = [], [] size, action_count = 0, 0 for action, data in actions: + raw_action = action action = serializer.dumps(action) cur_size = len(action) + 1 @@ -71,9 +72,9 @@ def _chunk_actions(actions, chunk_size, max_chunk_bytes, serializer): bulk_actions.append(action) if data is not None: bulk_actions.append(data) - bulk_data.append((action, data)) + bulk_data.append((raw_action, data)) else: - bulk_data.append((action, )) + bulk_data.append((raw_action, )) size += cur_size action_count += 1 @@ -237,7 +238,7 @@ def parallel_bulk(client, actions, thread_count=4, chunk_size=500, try: for result in pool.imap( - lambda bulk_data, bulk_actions: list(_process_bulk_chunk(client, bulk_actions, bulk_data, **kwargs)), + lambda bulk_chunk: list(_process_bulk_chunk(client, bulk_chunk[1], bulk_chunk[0], **kwargs)), _chunk_actions(actions, chunk_size, max_chunk_bytes, client.transport.serializer) ): for item in result: