Elasticsearch

Utilities for indexing data to Elasticsearch with Apache Beam.

class resiliparse.beam.elasticsearch.ElasticsearchBulkIndex(es_args, default_index=None, update=False, parallelism=None, buffer_size=3200, chunk_size=800, max_retries=10, initial_backoff=2, max_backoff=600, request_timeout=240, ignore_400=True, dry_run=False)

Bases: PTransform

Bulk-index documents to Elasticsearch.

Expects bulk index actions that can be passed to elasticsearch.helpers.streaming_bulk(). Helpers for creating such actions are given with index_action(), update_action(), and delete_action().

If the records are KV pairs, the key will be used as the document _id and the value is expected to be the document that is to be indexed as a dict. In this case, fields starting with underscores will be discarded and default_index must be set.

Returns the document IDs of successfully indexed documents.

Parameters:

  • es_args (Dict[<class 'str'>, Any]) – Elasticsearch client arguments

  • default_index (Union[<class 'NoneType'>, <class 'str'>]) – default index name if none is given in the index action

  • update (bool) – do a bulk UPDATE instead of a bulk INDEX (requires documents to exist)

  • parallelism (int) – reshuffle to achieve the desired level of parallelism

  • buffer_size (int) – internal buffer size

  • chunk_size (int) – indexing chunk size

  • max_retries (int) – maximum number of retries on recoverable failures

  • initial_backoff (float) – initial retry backoff

  • max_backoff (float) – maximum retry backoff

  • request_timeout (int) – Elasticsearch request timeout

  • ignore_400 (bool) – ignore HTTP 400 errors (skip unindexable documents)

  • dry_run (bool) – discard documents and do not actually index them

resiliparse.beam.elasticsearch.delete_action(doc_id, index)

Build a delete bulk action.

Parameters:

  • doc_id (str) – document ID

  • index (str) – index name

Returns:

index action dict

resiliparse.beam.elasticsearch.ensure_index(client, index_name, index_settings=None, mapping=None)

Helper function to ensure an index exists.

If the index does not exist, it will be created with the given mapping and settings.

Parameters:

  • client (Elasticsearch) – Elasticsearch client instance

  • index_name (str) – index name

  • index_settings (Dict[<class 'str'>, <class 'str'>]) – index settings dict

  • mapping (Dict[<class 'str'>, <class 'str'>]) – index mapping dict

resiliparse.beam.elasticsearch.index_action(doc_id, index, doc)

Build an index bulk action.

Parameters:

  • doc_id (str) – document ID (None for using autogenerated IDs)

  • index (Union[<class 'NoneType'>, <class 'str'>]) – index name

  • doc (Dict[<class 'str'>, Any]) – document as dict

Returns:

index action dict

resiliparse.beam.elasticsearch.update_action(doc_id, index, partial_doc)

Build an update bulk action.

Parameters:

  • doc_id (str) – document ID

  • index (str) – index name

  • partial_doc (Dict[<class 'str'>, Any]) – partial update document as dict

Returns:

index action dict