使用DBIC将多个"新"项插入数据库

Question

我正在进行一项生物信息学项目,该项目要求我从各种生物体中读取基因组数据(没有太多花哨,只需将其视为字符串)并将其插入数据库中.每次读取属于一个生物体,并且可以包含5000到5000万个基因,我需要在储存之前对其进行处理和分析.

目前执行此操作的脚本是用perl编写的,并且在完成所有计算之后,将结果存储在散列中,如下所示:

$new{$id}{gene_name}              = $id;
$new{$id}{gene_database_source} = $gene_database_source
$new{$id}{product}            = $product;
$new{$id}{sequence}               = $sequence;
$new{$id}{seqlength}              = $seqlength;
$new{$id}{digest}             = $digest;
$new{$id}{mw}                     = $mw;
$new{$id}{iep}                = $iep;
$new{$id}{tms}                = $tms;

在读取所有基因之后,插入将通过散列循环到eval {}语句中.

eval {
foreach my $id (keys %new) {

  my $rs = $schema->resultset('Genes')->create(
    {
        gene_name               => $new{$id}{gene_name},
        gene_product            => $new{$id}{product},
        sequence                => $new{$id}{sequence},
        gene_protein_length     => $new{$id}{seqlength},
        digest                  => $new{$id}{digest},
        gene_isoelectric_point  => $new{$id}{iep},
        gene_molecular_weight   => $new{$id}{mw},
        gene_tmd_count          => $new{$id}{tms},
        gene_species            => $species,
        species_code            => $spc,
        user_id                 => $tdruserid,
        gene_database_source    => $new{$id}{gene_database_source}

    }
  );
}; 

虽然这"有效",但它至少有两个我想解决的问题:

• eval语句旨在对插入进行"故障保护":如果其中一个插入失败,则eval将死亡并且不会进行插入.这显然不是eval的工作原理.我很确定在失败点之前所做的所有插入都将完成,并且无论如何都没有回滚.

• 脚本需要在非常大的数据集中循环两次(一次读取和创建哈希值,一次读取哈希值并执行插入时).这使得该过程的表现相当差.

我没有创建哈希,而是考虑使用DBIX 的新指令,$schema->new({..stuff..});然后进行大规模的插入事务.这将解决双重迭代,并且eval将与单个事务一起工作(或不工作),这将执行<所有插入或无>的预期行为... 有没有办法做到这一点？

Answer 1

您可以使用TxnScopeGuardDBIC创建大规模事务.在最基本的形式中,这将是如下.

eval { # or try from Try::Tiny
    my $guard = $schema->txn_scope_guard;

    foreach my $id ( keys %new ) {
        my $rs = $schema->resultset('Genes')->create(
            {
                gene_name              => $new{$id}{gene_name},
                gene_product           => $new{$id}{product},
                sequence               => $new{$id}{sequence},
                gene_protein_length    => $new{$id}{seqlength},
                digest                 => $new{$id}{digest},
                gene_isoelectric_point => $new{$id}{iep},
                gene_molecular_weight  => $new{$id}{mw},
                gene_tmd_count         => $new{$id}{tms},
                gene_species           => $species,
                species_code           => $spc,
                user_id                => $tdruserid,
                gene_database_source   => $new{$id}{gene_database_source}

            }
        );
    }
    $guard->commit;
}

您创建了一个范围保护对象,当您完成设置事务时,就可以commit了.如果对象超出范围,即因为某些东西died,它将自动回滚事务.

该eval能赶上die,你的程序将不会崩溃.你有那个部分是正确的,但你的代码也不会撤消以前的插入.需要注意的是尽量::微小的try提供更好的语法.但这里不需要它.

在这种情况下,事务意味着所有查询都被收集并同时运行.

请注意,这仍然INSERT只会为每个语句插入一行!

如果你想创建更大的INSERT语句,如下所示,你需要populate,而不是new.

INSERT INTO foo (bar, baz) VALUES
(1, 1),
(2, 2),
(3, 3),
...

该populate方法允许您一次传入一个包含多行的数组引用.这应该比一次插入一个更快.

$schema->resultset("Artist")->populate([
  [ qw( artistid name ) ],
  [ 100, 'A Formally Unknown Singer' ],
  [ 101, 'A singer that jumped the shark two albums ago' ],
  [ 102, 'An actually cool singer' ],
]);

转换为你的循环,如下所示.请注意,文档声称如果在void上下文中运行它会更快.

eval {
    $schema->resultset('Genes')->populate(
        [
            [
                                qw(
                    gene_name             gene_product   sequence
                    gene_protein_length   digest         gene_isoelectric_point
                    gene_molecular_weight gene_tmd_count gene_species
                    species_code          user_id        gene_database_source
        )
            ],
            map {
                [
                    $new{$_}{gene_name}, $new{$_}{product},
                    $new{$_}{sequence},  $new{$_}{seqlength},
                    $new{$_}{digest},    $new{$_}{iep},
                    $new{$_}{mw},        $new{$_}{tms},
                    $species,            $spc,
                    $tdruserid,          $new{$_}{gene_database_source},
                ]
            } keys %new
        ],
    );
}

像这样,不需要范围保护.但是,我建议你不要每个语句超过1000行.出于性能原因,以块的形式处理它可能是一个好主意.在这种情况下,您一次循环键1000.List :: MoreUtils有一个很好的natatime功能.

use List::MoreUtils 'natatime';

eval {
    my $guard = $schema->txn_scope_guard;

    my $it = natatime 1_000, keys %new;

    while ( my @keys = $it->() ) {
        $schema->resultset('Genes')->populate(
            [
                [
                    qw(
                        gene_name             gene_product   sequence
                        gene_protein_length   digest         gene_isoelectric_point
                        gene_molecular_weight gene_tmd_count gene_species
                        species_code          user_id        gene_database_source
                        )
                ],
                map {
                    [
                        $new{$_}{gene_name}, $new{$_}{product},
                        $new{$_}{sequence},  $new{$_}{seqlength},
                        $new{$_}{digest},    $new{$_}{iep},
                        $new{$_}{mw},        $new{$_}{tms},
                        $species,            $spc,
                        $tdruserid,          $new{$_}{gene_database_source},
                    ]
                } @keys
            ],
        );
    }

    $guard->commit;
}

现在它将每次插入1000行,并在一个大事务中运行所有这些查询.如果其中一个失败,则不会进行任何操作.

脚本需要在非常大的数据集中循环两次(一次读取和创建哈希值,一次读取哈希值并执行插入时).这使得该过程的表现相当差.

除了这项任务之外,您没有展示如何创建数据.

$new{$id}{gene_name}              = $id;
$new{$id}{gene_database_source} = $gene_database_source
$new{$id}{product}            = $product;

如果这就是它的全部,那么没有什么能阻止你直接使用我上面所示的方法,你在第一次处理数据并构建哈希时.以下代码不完整,因为您没有告诉我们数据的来源,但您应该获得要点.

eval {
    my $guard = $schema->txn_scope_guard;

    # we use this to collect rows to process
    my @rows;

    # this is where your data comes in
    while ( my $foo = <DATA> ) {

        # here you process the data and come up with your variables
        my ( $id, $gene_database_source, $product, $sequence, $seqlength, 
             $digest, $mw, $iep, $tms );

        # collect the row so we can insert it later
        push(
            @rows,
            [
                $id, $gene_database_source, $product, $sequence, $seqlength, 
                $digest, $mw, $iep, $tms,
            ]
        );

        # only insert if we reached the limit
        if ( scalar @rows == 1000 ) {
            $schema->resultset('Genes')->populate(
                [
                    [
                        qw(
                            gene_name             gene_product   sequence
                            gene_protein_length   digest         gene_isoelectric_point
                            gene_molecular_weight gene_tmd_count gene_species
                            species_code          user_id        gene_database_source
                            )
                    ],
                    \@rows,
                ],
            );

            # empty the list of values
            @rows = ();
        }
    }
    $guard->commit;
}

基本上我们在处理它们时直接收集多达1000行作为数组引用,当我们达到限制时,我们将它们传递给数据库.然后我们重置行数组并重新开始.同样,所有这些都包含在事务中,因此只有在所有插入都正常的情况下才会提交.

有关DBIC中交易的更多信息.

请注意,我尚未测试任何此代码.