如何用Perl的Search::Elasticsearch插入含嵌套结构的ES文档

我从SQL数据库提取数据，尝试使用Search::Elasticsearch插件将记录写入包含大量嵌套对象的Elasticsearch索引，测试发现以下两种写法的差异：

可行写法 #

$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 }]}});

不可行写法 #

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

显然bulk->create需要接收哈希或哈希引用格式的参数，而非字符串。我不确定是否有可行的字符串转换方法，或是存在更优的实现方式。若为扁平结构，可直接将数据推入数组后写入，但由于存在嵌套结构（每个记录可包含多个applications），需要循环处理相关信息。以下是我正在尝试的未完成代码示例（含硬编码的'13'等待完善部分）：

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;

内容的提问来源于stack exchange，提问作者markedperf