使用Perl DBI批量导入Snowflake提速方法咨询

我通过哈希存储需插入Snowflake表的2个字段值，总计约450万条数据。当前采用如下Perl DBI代码，通过execute_array每次批量插入16000条，但加载速度极慢，请问是否有更快的批量加载方案？

my $ccptable = "DWH_CCP.TGT.CCP_CNTC_BRDBAND_PING_DTL";
my ($sth, $sql);
$sql = "insert into $ccptable (op_hh_id,pstg_ts,load_ts) values (?, ?, current_timestamp)";
$sth = $dbh->prepare("$sql");

my (@col1, @col2);
my @status;
my $acctkeycnt = 0;
my $loopcnt = 0;
foreach my $acctkey (keys %allacctkeys) {
  my $timekey = $allacctkeys{$acctkey};
  $acctkeycnt++;
  push @col1, $acctkey;
  push @col2, $timekey;

  if ($acctkeycnt > 16000) {
    $loopcnt++;
    $sth->bind_param_array(1, \@col1);
    $sth->bind_param_array(2, \@col2);
    print "\nExecuting Array Load  Count => $loopcnt\n";
    $sth->execute_array( { ArrayTupleStatus => \@status } );
    @col1 = ();
    @col2 = ();
  }
}

1. 使用Snowflake原生COPY INTO（推荐） #

这是Snowflake批量加载数据的最优方案，效率远高于INSERT类批量操作，核心思路是先将数据导出为结构化文件，再通过Snowflake的COPY命令直接加载：

• 将哈希中的数据写入本地CSV/TSV文件，仅保留op_hh_id和pstg_ts两个字段（load_ts可在COPY阶段指定）
• 执行COPY INTO语句完成加载，示例代码：

# 写入临时CSV文件
open my $fh, '>', '/tmp/ping_data.csv' or die $!;
foreach my $acctkey (keys %allacctkeys) {
    my $timekey = $allacctkeys{$acctkey};
    print $fh "$acctkey,$timekey\n";
}
close $fh;

# 执行COPY加载
my $copy_sql = qq{
    COPY INTO $ccptable (op_hh_id, pstg_ts, load_ts)
    FROM '/tmp/ping_data.csv'
    FILE_FORMAT = (TYPE = CSV FIELD_OPTIONALLY_ENCLOSED_BY = '"' SKIP_HEADER = 0)
    SET = (load_ts = CURRENT_TIMESTAMP())
};
$dbh->do($copy_sql);

若数据量过大，可拆分多个文件并行加载，或结合Snowflake的外部阶段（Stage）存储进一步提升效率。

2. 优化execute_array的批量逻辑 #

如果暂时无法切换到COPY方案，可通过以下调整提升现有代码效率：

• 增大批量条数：当前16000条的批量可尝试提升至64000或128000（需注意Perl内存占用，避免数组过大导致内存溢出）
• 关闭自动提交：默认DBI自动提交会导致每次批量插入都触发事务，改为手动批量提交减少开销：

$dbh->{AutoCommit} = 0; # 关闭自动提交
my $commit_interval = 4; # 每4次批量提交一次事务

# ... 原有循环代码 ...
if ($acctkeycnt > 16000) {
    $loopcnt++;
    $sth->bind_param_array(1, \@col1);
    $sth->bind_param_array(2, \@col2);
    print "\nExecuting Array Load  Count => $loopcnt\n";
    $sth->execute_array( { ArrayTupleStatus => \@status } );
    @col1 = ();
    @col2 = ();

    # 达到提交间隔时执行提交
    if ($loopcnt % $commit_interval == 0) {
        $dbh->commit;
    }
}

# 循环结束后提交剩余数据
$dbh->commit;
$dbh->{AutoCommit} = 1; # 恢复自动提交

3. 预生成多行INSERT语句（应急方案） #

若上述方案都无法实施，可生成包含多行值的INSERT语句（每次1000-5000条），但需注意SQL语句长度限制，示例逻辑：

my $batch_size = 5000;
my @values;
foreach my $acctkey (keys %allacctkeys) {
    my $timekey = $allacctkeys{$acctkey};
    push @values, "('$acctkey', '$timekey', CURRENT_TIMESTAMP)";
    
    if (scalar @values >= $batch_size) {
        my $insert_sql = "INSERT INTO $ccptable VALUES " . join(',', @values);
        $dbh->do($insert_sql);
        @values = ();
    }
}
# 插入剩余数据
if (@values) {
    my $insert_sql = "INSERT INTO $ccptable VALUES " . join(',', @values);
    $dbh->do($insert_sql);
}

此方案效率不如前两种，仅作应急使用。

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