在我的例子中，问题是我从SQL数据库中提取数据，并尝试使用Search：：Elasticsearch插件将这些记录创建到Elasticsearch中，其中弹性模式充满了嵌套对象。作为一个例子，我查询SQL数据库，组合信息以构建格式并尝试写入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);

看起来信息需要以哈希或hashref格式而不是字符串格式。我不确定是否有一种方法可以转换字符串，或者是否有一种更好的方法。
如果它的模式是平面的，你可以直接把信息推到一个数组中，然后把它写出来。但是因为它包含嵌套的，所以可以有多个应用程序循环这些信息。
这里有一个草图，我试图做什么。没有完成，如硬编码的'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;