Helpers

Collection of simple helper functions that abstract some specifics or the raw API.

Bulk helpers

There are several helpers for the bulk API since it’s requirement for specific formatting and other considerations can make it cumbersome if used directly.

All bulk helpers accept an instance of Elasticsearch class and an iterable actions (any iterable, can also be a generator, which is ideal in most cases since it will allow you to index large datasets without the need of loading them into memory).

The items in the action iterable should be the documents we wish to index in several formats. The most common one is the same as returned by search(), for example:

{
    '_index': 'index-name',
    '_type': 'document',
    '_id': 42,
    '_parent': 5,
    'pipeline': 'my-ingest-pipeline',
    '_source': {
        "title": "Hello World!",
        "body": "..."
    }
}

Alternatively, if _source is not present, it will pop all metadata fields from the doc and use the rest as the document data:

{
    "_id": 42,
    "_parent": 5,
    "title": "Hello World!",
    "body": "..."
}

The bulk() api accepts index, create, delete, and update actions. Use the _op_type field to specify an action (_op_type defaults to index):

{
    '_op_type': 'delete',
    '_index': 'index-name',
    '_type': 'document',
    '_id': 42,
}
{
    '_op_type': 'update',
    '_index': 'index-name',
    '_type': 'document',
    '_id': 42,
    'doc': {'question': 'The life, universe and everything.'}
}

Note

When reading raw json strings from a file, you can also pass them in directly (without decoding to dicts first). In that case, however, you lose the ability to specify anything (index, type, even id) on a per-record basis, all documents will just be sent to elasticsearch to be indexed as-is.

elasticsearch.helpers.streaming_bulk(client, actions, chunk_size=500, max_chunk_bytes=104857600, raise_on_error=True, expand_action_callback=<function expand_action>, raise_on_exception=True, **kwargs)

Streaming bulk consumes actions from the iterable passed in and yields results per action. For non-streaming usecases use bulk() which is a wrapper around streaming bulk that returns summary information about the bulk operation once the entire input is consumed and sent.

Parameters:
  • client – instance of Elasticsearch to use
  • actions – iterable containing the actions to be executed
  • chunk_size – number of docs in one chunk sent to es (default: 500)
  • max_chunk_bytes – the maximum size of the request in bytes (default: 100MB)
  • raise_on_error – raise BulkIndexError containing errors (as .errors) from the execution of the last chunk when some occur. By default we raise.
  • raise_on_exception – if False then don’t propagate exceptions from call to bulk and just report the items that failed as failed.
  • expand_action_callback – callback executed on each action passed in, should return a tuple containing the action line and the data line (None if data line should be omitted).
elasticsearch.helpers.parallel_bulk(client, actions, thread_count=4, chunk_size=500, max_chunk_bytes=104857600, queue_size=4, expand_action_callback=<function expand_action>, **kwargs)

Parallel version of the bulk helper run in multiple threads at once.

Parameters:
  • client – instance of Elasticsearch to use
  • actions – iterator containing the actions
  • thread_count – size of the threadpool to use for the bulk requests
  • chunk_size – number of docs in one chunk sent to es (default: 500)
  • max_chunk_bytes – the maximum size of the request in bytes (default: 100MB)
  • raise_on_error – raise BulkIndexError containing errors (as .errors) from the execution of the last chunk when some occur. By default we raise.
  • raise_on_exception – if False then don’t propagate exceptions from call to bulk and just report the items that failed as failed.
  • expand_action_callback – callback executed on each action passed in, should return a tuple containing the action line and the data line (None if data line should be omitted).
  • queue_size – size of the task queue between the main thread (producing chunks to send) and the processing threads.
elasticsearch.helpers.bulk(client, actions, stats_only=False, **kwargs)

Helper for the bulk() api that provides a more human friendly interface - it consumes an iterator of actions and sends them to elasticsearch in chunks. It returns a tuple with summary information - number of successfully executed actions and either list of errors or number of errors if stats_only is set to True. Note that by default we raise a BulkIndexError when we encounter an error so options like stats_only only apply when raise_on_error is set to False.

See streaming_bulk() for more accepted parameters

Parameters:
  • client – instance of Elasticsearch to use
  • actions – iterator containing the actions
  • stats_only – if True only report number of successful/failed operations instead of just number of successful and a list of error responses

Any additional keyword arguments will be passed to streaming_bulk() which is used to execute the operation.

Scan

elasticsearch.helpers.scan(client, query=None, scroll=u'5m', raise_on_error=True, preserve_order=False, size=1000, request_timeout=None, clear_scroll=True, **kwargs)

Simple abstraction on top of the scroll() api - a simple iterator that yields all hits as returned by underlining scroll requests.

By default scan does not return results in any pre-determined order. To have a standard order in the returned documents (either by score or explicit sort definition) when scrolling, use preserve_order=True. This may be an expensive operation and will negate the performance benefits of using scan.

Parameters:
  • client – instance of Elasticsearch to use
  • query – body for the search() api
  • scroll – Specify how long a consistent view of the index should be maintained for scrolled search
  • raise_on_error – raises an exception (ScanError) if an error is encountered (some shards fail to execute). By default we raise.
  • preserve_order – don’t set the search_type to scan - this will cause the scroll to paginate with preserving the order. Note that this can be an extremely expensive operation and can easily lead to unpredictable results, use with caution.
  • size – size (per shard) of the batch send at each iteration.
  • request_timeout – explicit timeout for each call to scan
  • clear_scroll – explicitly calls delete on the scroll id via the clear scroll API at the end of the method on completion or error, defaults to true.

Any additional keyword arguments will be passed to the initial search() call:

scan(es,
    query={"query": {"match": {"title": "python"}}},
    index="orders-*",
    doc_type="books"
)

Reindex

elasticsearch.helpers.reindex(client, source_index, target_index, query=None, target_client=None, chunk_size=500, scroll=u'5m', scan_kwargs={}, bulk_kwargs={})

Reindex all documents from one index that satisfy a given query to another, potentially (if target_client is specified) on a different cluster. If you don’t specify the query you will reindex all the documents.

Since 2.3 a reindex() api is available as part of elasticsearch itself. It is recommended to use the api instead of this helper wherever possible. The helper is here mostly for backwards compatibility and for situations where more flexibility is needed.

Note

This helper doesn’t transfer mappings, just the data.

Parameters:
  • client – instance of Elasticsearch to use (for read if target_client is specified as well)
  • source_index – index (or list of indices) to read documents from
  • target_index – name of the index in the target cluster to populate
  • query – body for the search() api
  • target_client – optional, is specified will be used for writing (thus enabling reindex between clusters)
  • chunk_size – number of docs in one chunk sent to es (default: 500)
  • scroll – Specify how long a consistent view of the index should be maintained for scrolled search
  • scan_kwargs – additional kwargs to be passed to scan()
  • bulk_kwargs – additional kwargs to be passed to bulk()