How to insert create document into Elasticsearch using the Perl client that contains a nested schema

Question

In my case, the problem is I'm extracting data from a SQL database and trying to create those records into Elasticsearch using the Search::Elasticsearch plugin where the elastic schema is full of nested objects. As an example, I'm querying the SQL database, combining the information to build the format and trying to write to Elasticsearch.

This works:

$bulk->create({ id => 1, source => { applications => [ { source => $source, cpe => $cpe, firstseen => $firstseen, lastseen => $lastseen }, { source => $source2, cpe => $cpe2, firstseen => $firstseen2, lastseen => $lastseen2 }]}},{ id => 2, source => { applications => [ { source => $source, cpe => $cpe, firstseen => $firstseen, lastseen => $lastseen }, { source => $source2, cpe => $cpe2, firstseen => $firstseen2, lastseen => $lastseen2 }]}});

This does not work:

my $query = "{ id => 1, source => { applications => [ { source => $source, cpe => $cpe, firstseen => $firstseen, lastseen => $lastseen }, { source => $source2, cpe => $cpe2, firstseen => $firstseen2, lastseen => $lastseen2 }]}},{ id => 2, source => { applications => [ { source => $source, cpe => $cpe, firstseen => $firstseen, lastseen => $lastseen }, { source => $source2, cpe => $cpe2, firstseen => $firstseen2, lastseen => $lastseen2 }]}}";

$bulk->create($query);

It looks like the information needs to be in a hash or hashref format versus string. I'm not sure if there's a way to convert the string that would work or if a better way in general.

If it schema was flat, you can push the information directly to an array and then write it out. But since it contains nested, there can be more than one applications, looping through that information.

Here's a sketch of what i'm trying to do. Not finished, such as the hard coded '13'...but trying to work through the main issue...

my $e = Search::Elasticsearch->new( nodes => '192.168.1.11:9200' );
my $index_exists = $e->indices->exists( index => 'sql_software' );
if ($index_exists) { $e->indices->delete( index => 'sql_software' ); }
$e->indices->create( index => 'sql_software' );
my $bulk = $e->bulk_helper( index => 'sql_software');

my $dbh = DBI->connect("dbi:ODBC:Driver=/opt/microsoft/msodbcsql17/lib64/libmsodbcsql-17.10.so.4.1;Server=192.168.1.11;Database=Software;UID=sa;PWD=xxxx");

        my $sth = $dbh->prepare("SELECT FD.DeviceId AS 'idlist' FROM Software.Device FD");
        $sth->execute();
        my $list = $sth->fetchall_arrayref({});

        foreach my $res (@{ $list }) {
                $sth = $dbh->prepare("SELECT FCR.DeviceId, RT.ResourceTypeName AS 'source', DC.CpeDesc AS 'cpe', FCR.InstallDate AS 'firstSeen', FCR.LDate AS 'lastSeen' FROM [Software].[DimCpe] DC JOIN [Software].[CpeResults] FCR ON FCR.CpeId = DC.CpeId JOIN Software.DimResource DR ON DR.ResourceId = FCR.ResourceId JOIN Software.ResourceType RT ON RT.ResourceTypeId = DR.ResourceTypeId WHERE FCR.DeviceId = ? order by FCR.DeviceId");

                $sth->execute($res->{idlist}) or die $dbh->errstr;

                my $saveval = $res->{idlist};

                my $savesoftware="";
                my $i=0;
                while ( my @row = $sth->fetchrow_array ) {

                        my $deviceid = @row[0];
                        my $source = @row[1];
                        my $cpe = @row[2];
                        my $firstseen = @row[3];
                        my $lastseen = @row[4];

                        if ($i==0) {
                                $savesoftware = "{ id => $deviceid, source => { applications => [ ";
                        } elsif ($i==13) {
                                $savesoftware = $savesoftware . "{ source => $source, cpe => $cpe, firstseen => $firstseen, lastseen => $lastseen }]}}";
                        } else {
                                $savesoftware = $savesoftware . "{ source => $source, cpe => $cpe, firstseen => $firstseen, lastseen => $lastseen },";
                        }
                        $i++;
                }
                $bulk->create( $savesoftware );
        }

$sth->finish;
$bulk->flush;
$dbh->disconnect;

Answer 1

You could eval the string to a hash.

my $query = eval "{ id => 1, source => { applications => [ { source => $source, cpe => $cpe, firstseen => $firstseen, lastseen => $lastseen }, { source => $source2, cpe => $cpe2, firstseen => $firstseen2, lastseen => $lastseen2 }]}},{ id => 2, source => { applications => [ { source => $source, cpe => $cpe, firstseen => $firstseen, lastseen => $lastseen }, { source => $source2, cpe => $cpe2, firstseen => $firstseen2, lastseen => $lastseen2 }]}}";

$bulk->create($query);

But I would probably just refactor your code to manipulate the hash instead of making a string, the more idiomatic Perl way to do it would be something like this:

my %item;

$item{applications} = [];

my $i=0;
while ( my @row = $sth->fetchrow_array ) {
    my $deviceid = $row[0];
    my $source = $row[1];
    my $cpe = $row[2];
    my $firstseen = $row[3];
    my $lastseen = $row[4];

    if ($i==0) {
        $item{deviceid} = $row[0];
    } else {
        push @{$item{applications}}, { source => $source, cpe => $cpe, firstseen => $firstseen, lastseen => $lastseen };
    }

    $i++;
}

$bulk->create(\%item);