Elasticsearch::Bulk - A helper module for the Bulk API and for reindexing
version 1.04
use Elasticsearch; use Elasticsearch::Bulk; my $es = Elasticsearch->new; my $bulk = Elasticsearch::Bulk->new( es => $es, index => 'my_index', type => 'my_type' ); # Index docs: $bulk->index({ id => 1, source => { foo => 'bar' }}); $bulk->add_action( index => { id => 1, source => { foo=> 'bar' }}); # Create docs: $bulk->create({ id => 1, source => { foo => 'bar' }}); $bulk->add_action( create => { id => 1, source => { foo=> 'bar' }}); $bulk->create_docs({ foo => 'bar' }) # Delete docs: $bulk->delete({ id => 1}); $bulk->add_action( delete => { id => 1 }); $bulk->delete_ids(1,2,3) # Update docs: $bulk->update({ id => 1, script => '...' }); $bulk->add_action( update => { id => 1, script => '...' }); # Manual flush $bulk->flush # Reindex docs: $bulk = Elasticsearch::Bulk->new( es => $es, index => 'new_index', verbose => 1 ); $bulk->reindex( source => { index => 'old_index' });
This module provides a wrapper for the "bulk()" in Elasticsearch::Client::Direct method which makes it easier to run multiple create, index, update or delete actions in a single request. It also provides a simple interface for reindexing documents.
The Elasticsearch::Bulk module acts as a queue, buffering up actions until it reaches a maximum count of actions, or a maximum size of JSON request body, at which point it issues a bulk() request.
bulk()
Once you have finished adding actions, call "flush()" to force the final bulk() request on the items left in the queue.
This class does Elasticsearch::Role::Bulk and Elasticsearch::Role::Is_Sync.
new()
$bulk = Elasticsearch::Bulk->new( es => $es, # required index => 'default_index', # optional type => 'default_type', # optional %other_bulk_params # optional max_count => 1_000, # optional max_size => 1_000_000, # optional verbose => 0 | 1, # optional on_success => sub {...}, # optional on_error => sub {...}, # optional on_conflict => sub {...}, # optional );
The new() method returns a new $bulk object. You must pass your Elasticsearch client as the es argument.
$bulk
es
The index and type parameters provide default values for index and type, which can be overridden in each action. You can also pass any other values which are accepted by the bulk() method.
index
type
See "flush()" for more information about the other parameters.
flush()
$result = $bulk->flush;
The flush() method sends all buffered actions to Elasticsearch using a bulk() request.
An automatic "flush()" is triggered whenever the max_count or max_size threshold is breached. This causes all actions in the buffer to be sent to Elasticsearch.
max_count
max_size
The maximum number of actions to allow before triggering a "flush()". This can be disabled by setting max_count to 0. Defaults to 1,000.
0
1,000
The maximum size of JSON request body to allow before triggering a "flush()". This can be disabled by setting max_size to 0. Defaults to 1_000,000 bytes.
1_000,000
There are three levels of error which can be thrown when "flush()" is called, either manually or automatically.
Temporary Elasticsearch errors
For instance, a NoNodes error which indicates that your cluster is down. These errors do not clear the buffer, as they can be retried later on.
NoNodes
Request errors
For instance, if one of your actions is malformed (eg you are missing a required parameter like index) then the whole "flush()" request is aborted and the buffer is cleared of all actions.
Action errors
Individual actions may fail. For instance, a create action will fail if a document with the same index, type and id already exists. These action errors are reported via callbacks.
create
id
By default, any Action errors (see above) cause warnings to be written to STDERR. However, you can use the on_error, on_conflict and on_success callbacks for more fine-grained control.
STDERR
on_error
on_conflict
on_success
All callbacks receive the following arguments:
$action
The name of the action, ie index, create, update or delete.
update
delete
$response
The response that Elasticsearch returned for this action.
$i
The index of the action, ie the first action in the flush request will have $i set to 0, the second will have $i set to 1 etc.
1
$bulk = Elasticsearch->new( es => $es, on_success => sub { my ($action,$response,$i) = @_; # do something }, );
The on_success callback is called for every action that has a successful response.
$bulk = Elasticsearch::Bulk->new( es => $es, on_conflict => sub { my ($action,$response,$i,$version) = @_; # do something }, );
The on_conflict callback is called for actions that have triggered a Conflict error, eg trying to create a document which already exists. The $version argument will contain the version number of the document currently stored in Elasticsearch (if found).
Conflict
$version
$bulk = Elasticsearch::Bulk->new( es => $es, on_error => sub { my ($action,$response,$i) = @_; # do something }, );
The on_error callback is called for any error (unless the on_conflict) callback has already been called).
If you want to be in control of flushing, and you just want to receive the raw response that Elasticsearch sends instead of using callbacks, then you can do so as follows:
$bulk = Elasticsearch::Bulk->new( es => $es, max_count => 0, max_size => 0, on_error => undef ); $bulk->add_actions(....); $response = $bulk->flush;
add_action()
$bulk->add_action( create => { ...params... }, index => { ...params... }, update => { ...params... }, delete => { ...params... } );
The add_action() method allows you to add multiple create, index, update and delete actions to the queue. The first value is the action type, and the second value is the parameters that describe that action. See the individual helper methods below for details.
Note: Parameters like index or type can be specified as index or as _index, so the following two lines are equivalent:
_index
index => { index => 'index', type => 'type', id => 1, source => {...}}, index => { _index => 'index', _type => 'type', _id => 1, _source => {...}},
Note: The index and type parameters can be specified in the params for any action, but if not specified, will default to the index and type values specified in "new()". These are required parameters: they must be specified either in "new()" or in every action.
create()
$bulk->create( { index => 'custom_index', source => { doc body }}, { type => 'custom_type', id => 1, source => { doc body }}, ... );
The create() helper method allows you to add multiple create actions. It accepts the same parameters as "create()" in Elasticsearch::Client::Direct except that the document body should be passed as the source or _source parameter, instead of as body.
source
_source
body
create_docs()
$bulk->create_docs( { doc body }, { doc body }, ... );
The create_docs() helper is a shorter form of "create()" which can be used when you are using the default index and type as set in "new()" and you are not specifying a custom id per document. In this case, you can just pass the individual document bodies.
index()
$bulk->index( { index => 'custom_index', source => { doc body }}, { type => 'custom_type', id => 1, source => { doc body }}, ... );
The index() helper method allows you to add multiple index actions. It accepts the same parameters as "index()" in Elasticsearch::Client::Direct except that the document body should be passed as the source or _source parameter, instead of as body.
delete()
$bulk->delete( { index => 'custom_index', id => 1}, { type => 'custom_type', id => 2}, ... );
The delete() helper method allows you to add multiple delete actions. It accepts the same parameters as "delete()" in Elasticsearch::Client::Direct.
delete_ids()
$bulk->delete_ids(1,2,3...)
The delete_ids() helper method can be used when all of the documents you want to delete have the default index and type as set in "new()". In this case, all you have to do is to pass in a list of IDs.
update()
$bulk->update( { id => 1, doc => { partial doc }, doc_as_upsert => 1 }, { id => 2, lang => 'mvel', script => '_ctx.source.counter+=incr', params => { incr => 1}, upsert => { upsert doc } }, ... );
The update() helper method allows you to add multiple update actions. It accepts the same parameters as "update()" in Elasticsearch::Client::Direct. An update can either use a partial doc which gets merged with an existing doc (example 1 above), or can use a script to update an existing doc (example 2 above).
script
A common use case for bulk indexing is to reindex a whole index when changing the type mappings or analysis chain. This typically combines bulk indexing with scrolled searches: the scrolled search pulls all of the data from the source index, and the bulk indexer indexes the data into the new index.
reindex()
$bulk->reindex( source => $source, # required transform => \&transform, # optional version_type => 'external|internal', # optional );
The reindex() method requires a $source parameter, which provides the source for the documents which are to be reindexed.
$source
If the source argument is a HASH ref, then the hash is passed to "new()" in Elasticsearch::Scroll to create a new scrolled search.
$bulk = Elasticsearch::Bulk->new( index => 'new_index', verbose => 1 ); $bulk->reindex( source => { index => 'old_index', size => 500, # default search_type => 'scan' # default } );
If a default index or type has been specified in the call to "new()", then it will replace the index and type values for the docs returned from the scrolled search. In the example above, all docs will be retrieved from "old_index" and will be bulk indexed into "new_index".
"old_index"
"new_index"
The source parameter also accepts a coderef or an anonymous sub, which should return one or more new documents every time it is executed. This allows you to pass any iterator, wrapped in an anonymous sub:
my $iter = get_iterator_from_somewhere(); $bulk->reindex( source => sub { $iter->next } );
The transform parameter allows you to change documents on the fly, using a callback. The callback receives the document as the only argument, and should return the updated document, or undef if the document should not be indexed:
transform
undef
$bulk->reindex( source => { index => 'old_index' }, transform => sub { my $doc = shift; # don't index doc marked as valid:false return undef unless $doc->{_source}{valid}; # convert $tag to @tags $doc->{_source}{tags} = [ delete $doc->{_source}{tag}]; return $doc } );
By default, "reindex()" expects the source and destination indices to be in the same cluster. To pull data from one cluster and index it into another, you can use two separate $es objects:
$es
$es_local = Elasticsearch->new( nodes => 'localhost:9200' ); $es_remote = Elasticsearch->new( nodes => 'search1:9200' ); Elasticsearch::Bulk->new( es => $es_local, verbose => 1 ) -> reindex( es => $es_remote );
If you are using parent-child relationships or custom routing values, and you want to preserve these when you reindex your documents, then you will need to request these values specifically, as follows:
routing
$bulk->reindex( source => { index => 'old_index', fields => ['_source','_parent','_routing'] } );
Every document in Elasticsearch has a current version number, which is used for optimistic concurrency control, that is, to ensure that you don't overwrite changes that have been made by another process.
version
All CRUD operations accept a version parameter and a version_type parameter which tells Elasticsearch that the change should only be made if the current document corresponds to these parameters. The version_type parameter can have the following values:
version_type
internal
Use Elasticsearch version numbers. Documents are only changed if the document in Elasticsearch has the same version number that is specified in the CRUD operation. After the change, the new version number is version+1.
version+1
external
Use an external versioning system, such as timestamps or version numbers from an external database. Documents are only changed if the document in Elasticsearch has a lower version number than the one specified in the CRUD operation. After the change, the new version number is version.
If you would like to reindex documents from one index to another, preserving the version numbers from the original index, then you need the following:
$bulk->reindex( source => { index => 'old_index', version => 1, # retrieve version numbers in search }, version_type => 'external' # use these "external" version numbers );
Clinton Gormley <drtech@cpan.org>
This software is Copyright (c) 2014 by Elasticsearch BV.
This is free software, licensed under:
The Apache License, Version 2.0, January 2004
To install Elasticsearch, copy and paste the appropriate command in to your terminal.
cpanm
cpanm Elasticsearch
CPAN shell
perl -MCPAN -e shell install Elasticsearch
For more information on module installation, please visit the detailed CPAN module installation guide.