NAME

Search::Elasticsearch::Client::8_0::Async::Bulk - A helper module for the Bulk API

VERSION

version 8.12

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;

DESCRIPTION

This module provides an async wrapper for the "bulk()" in Search::Elasticsearch::Client::8_0::Direct method which makes it easier to run multiple create, index, update or delete actions in a single request.

The Search::Elasticsearch::Client::8_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::8_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    => 6,                   # 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::8_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 to 0. Defaults to 1,000.
max_size

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.
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 a NoNodes 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 the on_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 same index, type and id 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 or 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.

`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::8_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::8_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::8_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::8_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::8_0::Direct.

AUTHOR

Enrico Zimuel <enrico.zimuel@elastic.co>

COPYRIGHT AND LICENSE

This is free software, licensed under:

The Apache License, Version 2.0, January 2004

To install Search::Elasticsearch::Client::8_0::Async, copy and paste the appropriate command in to your terminal.

cpanm

cpanm Search::Elasticsearch::Client::8_0::Async

CPAN shell

perl -MCPAN -e shell
install Search::Elasticsearch::Client::8_0::Async

For more information on module installation, please visit the detailed CPAN module installation guide.

	Global
`s`	Focus search bar
`?`	Bring up this help dialog

	GitHub
`g` `p`	Go to pull requests
`g` `i`	go to github issues (only if github is preferred repository)

	POD
`g` `a`	Go to author
`g` `c`	Go to changes
`g` `i`	Go to issues
`g` `d`	Go to dist
`g` `r`	Go to repository/SCM
`g` `s`	Go to source
`g` `b`	Go to file browse

	Search terms
module: (e.g. module:Plugin)
distribution: (e.g. distribution:Dancer auth)
author: (e.g. author:SONGMU Redis)
version: (e.g. version:1.00)

NAME

VERSION

SYNOPSIS

DESCRIPTION

CREATING A NEW INSTANCE

new()

FLUSHING THE BUFFER

flush()

Auto-flushing

Errors when flushing

Using callbacks

on_success

on_conflict

on_error

Disabling callbacks and autoflush

CREATE, INDEX, UPDATE, DELETE

add_action()

create()

create_docs()

index()

delete()

delete_ids()

update()