NAME
Search::Elasticsearch::Client::2_0::Async::Bulk - A helper module for the Bulk API and for reindexing
VERSION
version 5.00
SYNOPSIS
use Search::Elasticsearch::Async;
my $es = Search::Elasticsearch::Async->new;
my $bulk = $es->bulk_helper(
index => 'my_index',
type => 'my_type'
);
# Index docs:
$promise = $bulk->index({ id => 1, source => { foo => 'bar' }});
$promise = $bulk->add_action( index => { id => 1, source => { foo=> 'bar' }});
# Create docs:
$promise = $bulk->create({ id => 1, source => { foo => 'bar' }});
$promise = $bulk->add_action( create => { id => 1, source => { foo=> 'bar' }});
$promise = $bulk->create_docs({ foo => 'bar' })
# Delete docs:
$promise = $bulk->delete({ id => 1});
$promise = $bulk->add_action( delete => { id => 1 });
$promise = $bulk->delete_ids(1,2,3)
# Update docs:
$promise = $bulk->update({ id => 1, script => '...' });
$promise = $bulk->add_action( update => { id => 1, script => '...' });
# Manual flush
$promise = $bulk->flush;
# Reindex docs:
$bulk = $es->bulk_helper(
index => 'new_index',
verbose => 1
);
$promise = $bulk->reindex( source => { index => 'old_index' });
DESCRIPTION
This module provides an async wrapper for the "bulk()" in Search::Elasticsearch::Client::2_0::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 Search::Elasticsearch::Client::2_0::Async::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.
Once you have finished adding actions, call "flush()" to force the final bulk()
request on the items left in the queue.
This class does Search::Elasticsearch::Client::2_0::Role::Bulk and Search::Elasticsearch::Role::Is_Async.
CREATING A NEW INSTANCE
new()
$bulk = $es->bulk_helper(
index => 'default_index', # optional
type => 'default_type', # optional
%other_bulk_params # optional
max_count => 1_000, # optional
max_size => 1_000_000, # optional
max_time => 5, # optional
verbose => 0 | 1, # optional
on_success => sub {...}, # optional
on_error => sub {...}, # optional
on_conflict => sub {...}, # optional
on_fatal => sub {...}, # optional
);
The bulk_helper
method loads Search::Elasticsearch::Client::2_0::Async::Bulk, calls "new()" with the specified parameters and returns a new $bulk
object.
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.
See "flush()" for more information about the other parameters.
FLUSHING THE BUFFER
flush()
$promise = $bulk->flush;
The flush()
method sends all buffered actions to Elasticsearch using a bulk() request and returns a Promise, which is rejected if the bulk request fails or if any of the on_success
, on_error
or on_conflict
callbacks throws an exception, otherwise it is resolved with the items that have been flushed.
Auto-flushing
An automatic "flush()" is triggered whenever the max_count
, max_size
, or max_time
threshold is breached. This causes all actions in the buffer to be sent to Elasticsearch.
max_count
The maximum number of actions to allow before triggering a "flush()". This can be disabled by setting
max_count
to0
. Defaults to1,000
.max_size
The maximum size of JSON request body to allow before triggering a "flush()". This can be disabled by setting
max_size
to0
. Defaults to1_000,000
bytes.max_time
The maximum number of seconds to wait before triggering a flush. Defaults to
0
seconds, which means that it is disabled. Note: This timeout is only triggered when new items are added to the queue, not in the background.
Errors when flushing
There are two levels of error which can be thrown when "flush()" is called, either manually or automatically.
Temporary Elasticsearch errors
A
Cxn
error like aNoNodes
error which indicates that your cluster is down. These errors do not clear the buffer, as they can be retried later on. These errors are reported via theon_fatal
callback and by rejecting the promise returned by "flush()", "index()" etc.Action errors
Individual actions may fail. For instance, a
create
action will fail if a document with the sameindex
,type
andid
already exists. These action errors are reported via callbacks.
Using callbacks
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.
All callbacks receive the following arguments:
$action
-
The name of the action, ie
index
,create
,update
ordelete
. $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 to0
, the second will have$i
set to1
etc.
on_success
$bulk = $e->bulk_helper->new(
on_success => sub {
my ($action,$response,$i) = @_;
# do something
},
);
The on_success
callback is called for every action that has a successful response.
on_conflict
$bulk = $e->bulk_helper->new(
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).
on_error
$bulk = $e->bulk_helper->new(
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).
Disabling callbacks and autoflush
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 = $e->bulk_helper->new(
max_count => 0,
max_size => 0,
on_error => undef
);
$bulk->add_actions(....);
$bulk->flush
->then(
sub { my $response = shift; ...},
sub { my $error = shift; ....}
)
CREATE, INDEX, UPDATE, DELETE
The "add_action()", "create()", "create_docs()", "index()", "delete()", "delete_ids()" and "update()" methods all return a Promise, which is resolved once the actions have been added to the queue and AFTER the queue has been flushed (if necessary). It is important to wait for the promise to be resolved before continuing to queue more items, otherwise the pending requests may fill up your available memory.
For instance:
use Promises qw(deferred);
use Scalar::Util qw(weaken);
$bulk = $es->bulk_helper;
sub bulk_index {
my $d = deferred;
my $weak_cb;
my $cb = sub {
my @docs = get_next_docs_from_somewhere();
unless (@docs) {
return $d->resolve;
}
$bulk->index(@docs)
->then(
$weak_cb,
sub { $d->reject(@_) }
);
};
weaken ($weak_cb = $cb);
$cb->();
$d->promise->then( sub {$b->flush} );
}
add_action()
$promise = $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', 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()
$promise = $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 Search::Elasticsearch::Client::2_0::Direct except that the document body should be passed as the source
or _source
parameter, instead of as body
.
create_docs()
$promise = $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()
$promise = $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 Search::Elasticsearch::Client::2_0::Direct except that the document body should be passed as the source
or _source
parameter, instead of as body
.
delete()
$promise = $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 Search::Elasticsearch::Client::2_0::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()
$promise = $bulk->update(
{ id => 1,
doc => { partial doc },
doc_as_upsert => 1
},
{ id => 2,
script => { script },
upsert => { upsert doc }
},
...
);
The update()
helper method allows you to add multiple update
actions. It accepts the same parameters as "update()" in Search::Elasticsearch::Client::2_0::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). More information on script
can be found here: "update()" in Search::Elasticsearch::Client::2_0::Direct.
REINDEXING DOCUMENTS
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()
$promise = $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.
Reindexing from another index
If the source
argument is a HASH ref, then the hash is passed to "new()" in Search::Elasticsearch::Client::2_0::Async::Scroll to create a new scrolled search.
$bulk = $es->bulk_helper->new(
index => 'new_index',
verbose => 1
);
$bulk->reindex(
source => {
index => 'old_index',
size => 500, # default
search_type => 'scan' # default
}
)->then(
sub { say "Done"},
sub { die @_ }
);
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"
.
Reindexing from a generic source
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();
$promise = $bulk->reindex(
source => sub { $iter->next }
);
Transforming docs on the fly
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:
$promise = $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
}
);
Reindexing from another cluster
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_target = Search::Elasticsearch::Async->new( nodes => 'localhost:9200' );
$es_source = Search::Elasticsearch::Async->new( nodes => 'search1:9200' );
$promise = $es_target->bulk_helper(
verbose => 1
)
-> reindex(
source => {
es => $es_source,
index => 'my_index'
}
);
Parents and routing
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:
$promise = $bulk->reindex(
source => {
index => 'old_index',
fields => ['_source','_parent','_routing']
}
);
Working with version numbers
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.
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:
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 isversion+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 isversion
.
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:
$promise = $bulk->reindex(
source => {
index => 'old_index',
version => 1, # retrieve version numbers in search
},
version_type => 'external' # use these "external" version numbers
);
AUTHOR
Clinton Gormley <drtech@cpan.org>
COPYRIGHT AND LICENSE
This software is Copyright (c) 2016 by Elasticsearch BV.
This is free software, licensed under:
The Apache License, Version 2.0, January 2004