Search...::Bulk(3pm)

💖 Search...::Bulk(3pm) 💖

okay bestie, so this is the manual page for Search::Elasticsearch::Client::1_0::Bulk — part of the User Contributed Perl Documentation collection, and it is genuinely one of the most useful little helpers in the whole Elasticsearch Perl ecosystem. we're doing part 1 of 2 here, so buckle up 🎀

---

✨ NAME ✨

Search::Elasticsearch::Client::1_0::Bulk — a helper module for the Bulk API and for reindexing. it's giving "your personal assistant who batches stuff up so you don't have to" 💅

✨ VERSION ✨

version 6.81 — yes, this is the version, no cap 🌸

✨ SYNOPSIS ✨

okay so here's the whole vibe in one go, look at her go:

use Search::Elasticsearch;

my $es   = Search::Elasticsearch->new;
my $bulk = $es->bulk_helper(
    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:
my $bulk = $es->bulk_helper(
    index   => 'new_index',
    verbose => 1
);

$bulk->reindex( source => { index => 'old_index'});

like… she does everything. iconic behavior honestly 👑

✨ DESCRIPTION ✨

so this module is a wrapper around the "bulk()" in Search::Elasticsearch::Client::1_0::Direct method, and what it does is make it way easier to run a bunch of create, index, update or delete actions inside a single request. it also gives you a cute lil simple interface for reindexing documents. 🦋

Search::Elasticsearch::Client::1_0::Bulk literally acts like a queue — it buffers up all your actions until it hits a max count of actions, or a max size of JSON request body, and then it fires off the whole "bulk()" request. the way it just… holds them for you? obsessed.

once you're done adding actions, you call "flush()" to force that final "bulk()" request on whatever's left chillin' in the queue. don't forget her, she's important 💖

also fyi, this class does Search::Elasticsearch::Client::1_0::Role::Bulk and Search::Elasticsearch::Role::Is_Sync. just so you know where she gets it from 🫶

🌈 CREATING A NEW INSTANCE 🌈

"new()"

my $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

);

the "new()" method returns a fresh new $bulk object, slay. you must pass your Search::Elasticsearch client in as the "es" argument — non-negotiable bestie, she needs her client.

the "index" and "type" parameters set up default values for "index" and "type", and those can be overridden in each individual action if you want. you can also pass literally any other values that the bulk() method accepts, so go wild 💅

for the tea on all the other parameters, check out "flush()" below — she explains everything.

🧚 FLUSHING THE BUFFER 🧚

"flush()"

$result = $bulk->flush;

the "flush()" method sends all the buffered actions to Elasticsearch using a bulk() request. boom. done. ✨

Auto-flushing 🔥

an automatic "flush()" gets triggered any time the "max_count", "max_size", or "max_time" threshold gets breached. when that happens, every single action in the buffer gets sent off to Elasticsearch. no stragglers left behind 💖

• "max_count" — the max number of actions allowed before it triggers a "flush()". you can turn this off entirely by setting "max_count" to 0. default is "1,000".
• "max_size" — the max size of the JSON request body allowed before it triggers a "flush()". you can disable it by setting "max_size" to 0. default is "1_000,000" bytes. (yes that's a lot, she's got range 😌)
• "max_time" — the max number of seconds to wait before triggering a flush. defaults to 0 seconds, which means it's disabled. important note: this timeout only fires when new items are added to the queue — it does not run in the background. so don't be out here expecting her to wake up on her own, she won't 🙈

Errors when flushing 😭

there are two types of error that can be thrown when "flush()" is called — whether it's manual or automatic. let's get into it:

• Temporary Elasticsearch errors — a "Cxn" error, like a "NoNodes" error, which means your cluster is down. bestie… these errors do not clear the buffer, because you can retry them later on. so it's giving "temporary setback, not a full breakup" 💔➡️💖
• Action errors — individual actions can just fail, it happens. for example, a "create" action will fail if a document with the same "index", "type" and "id" already exists. these action errors get reported via callbacks.

Using callbacks 📞

by default, any Action errors (yeah, the ones we just talked about) cause warnings to be written to "STDERR". but if you want more fine-grained control, you can use the "on_error", "on_conflict" and "on_success" callbacks. so much power in your hands, use it responsibly 💅

all of the callbacks receive these arguments:

• $action — the name of the action, i.e. "index", "create", "update" or "delete".
• $response — the response that Elasticsearch returned for this action.
• $i — the index of the action, i.e. the first action in the flush request will have $i set to 0, the second will have $i set to 1, etc.

"on_success" 🎀

my $bulk = $es->bulk_helper(
    on_success  => sub {
        my ($action,$response,$i) = @_;
        # do something
    },
);

the "on_success" callback gets called for every single action that has a successful response. we love a win 🥳

"on_conflict" ⚔️

my $bulk = $es->bulk_helper(
    on_conflict  => sub {
        my ($action,$response,$i,$version) = @_;
        # do something
    },
);

the "on_conflict" callback gets called for actions that have triggered a "Conflict" error — for example, trying to "create" a document that already exists. awkward 😬 the $version argument will contain the version number of the document currently stored in Elasticsearch (if it was found).

"on_error" 💀

my $bulk = $es->bulk_helper(
    on_error  => sub {
        my ($action,$response,$i) = @_;
        # do something
    },
);

the "on_error" callback gets called for any error — unless the "on_conflict" callback has already been called first. so she's the backup, not the first responder 🚨

Disabling callbacks and autoflush 🙅♀️

if you wanna be fully in control of flushing, and you just want to receive the raw response that Elasticsearch sends instead of dealing with callbacks, then you can do it like this:

my $bulk = $es->bulk_helper(
    max_count   => 0,
    max_size    => 0,
    on_error    => undef
);

$bulk->add_actions(....);
$response = $bulk->flush;

look at you, taking the wheel. main character energy fr 🚗👑

👑 CREATE, INDEX, UPDATE, DELETE 👑

"add_action()"

$bulk->add_action(
    create => { ...params... },
    index  => { ...params... },
    update => { ...params... },
    delete => { ...params... }
);

the "add_action()" method lets you 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. for details, look at the individual helper methods below 💖

Note: parameters like "index" or "type" can be specified as "index" or as "_index" — so these two lines are literally 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 you don't specify them, they default to the "index" and "type" values you gave in "new()". and heads up — these are required parameters. they must be specified either in "new()" or in every single action. no skipping, bestie 🚫

"create()"

$bulk->create(
    { index => 'custom_index',         source => { doc body }},
    { type  => 'custom_type', id => 1, source => { doc body }},
    ...
);

the "create()" helper method lets you add multiple "create" actions. it accepts the same parameters as "create()" in Search::Elasticsearch::Client::1_0::Direct, except that the document body should be passed as the "source" or "_source" parameter instead of as "body". little swap, big difference ✨

"create_docs()"

$bulk->create_docs(
    { doc body },
    { doc body },
    ...
);

the "create_docs()" helper is a shorter form of "create()" — use her when you're using the default "index" and "type" set in "new()" and you're not specifying a custom "id" per document. in that case you can literally just pass the individual document bodies and vibe 🌸

"index()"

$bulk->index(
    { index => 'custom_index',         source => { doc body }},
    { type  => 'custom_type', id => 1, source => { doc body }},
    ...
);

the "index()" helper method lets you add multiple "index" actions. it accepts the same parameters as "index()" in Search::Elasticsearch::Client::1_0::Direct, except that the document body should be passed as the "source" or "_source" parameter instead of as "body". same energy as "create()", different action 💅

"delete()"

$bulk->delete(
    { index => 'custom_index', id => 1},
    { type  => 'custom_type',  id => 2},
    ...
);

the "delete()" helper method lets you add multiple "delete" actions. it accepts the same parameters as "delete()" in Search::Elasticsearch::Client::1_0::Direct. simple, clean, does the job 🗑️✨

"delete_ids()"

$bulk->delete_ids(1,2,3...)

the "delete_ids()" helper method is for when all the documents you wanna delete have the default "index" and "type" set in "new()". if that's the case, all you have to do is pass in a list of IDs and she handles the rest. so easy it's almost criminal 😌

"update()"

$bulk->update(
    { id            => 1,
      doc           => { partial doc },
      doc_as_upsert => 1
    },
    { id            => 2,
      lang          => 'mvel',
      script        => { script }
      upsert        => { upsert doc }
    },
    ...
);

the "update()" helper method lets you add multiple "update" actions. it accepts the same parameters as "update()" in Search::Elasticsearch::Client::1_0::Direct. an update can either use a partial doc that gets merged with an existing doc (that's example 1 up top 👆), or it can use a "script" to update an existing doc (that's example 2 ✌️). more info on "script" lives here: "update()" in Search::Elasticsearch::Client::1_0::Direct.

---

🦋 REINDEXING DOCUMENTS 🦋

okay bestie, so a super common glow-up for bulk indexing is reindexing a whole index when you're changing the type mappings or the analysis chain. It's giving "move out of the old place, into the new place." Usually you pair bulk indexing with scrolled searches: the scrolled search pulls all 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. No source, no slay.

🌸 Reindexing from another index 🌸

If the source argument is a HASH ref, then the hash is passed to new() in Search::Elasticsearch::Client::1_0::Scroll to create a new scrolled search. iconic behavior.

my $bulk = $es->bulk_helper(
    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. periodt.

🧚 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();

$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:

$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->new( nodes => 'localhost:9200' );
$es_source  = Search::Elasticsearch->new( nodes => 'search1:9200' );

my $bulk = $es_targert->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:

$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 (http://en.wikipedia.org/wiki/Optimistic_concurrency_control), that is, to ensure that you don't overwrite changes that have been made by another process. do not skip this part bestie 😭

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 is 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
);

🎀 AUTHOR 🎀

Enrico Zimuel <enrico.zimuel@elastic.co> — absolute legend 💖

🌈 COPYRIGHT AND LICENSE 🌈

This software is Copyright (c) 2020 by Elasticsearch BV.

This is free software, licensed under:

The Apache License, Version 2.0, January 2004

---

and that's the whole vibe, bestie 🫶 this page is from Search...::Bulk(3pm), perl v5.36.0, dated 2022-08-28. slay, hydrate, and may your reindexes always transform cleanly ✨💅