在5分钟内处理150万行文件

Question

经过网上长时间的搜索,我决定在这里询问我的问题.我有一个CSV文件集(总共36个文件),每5分钟一次.每个文件包含大约150万行.我需要在5分钟内处理这些文件.我必须解析这些文件并在存储区域内从它们创建所需的目录.然后将每个唯一行转换为文件并放入相关目录中.相关行也会写在相关文件中.如您所见,有许多I/O操作.

我可以完成12个文件大约10分钟.目标是在5分钟内完成36.我正在使用PERL来完成此操作.我看到的问题是系统调用i/o操作.

我想在Perl中控制文件处理程序和I/O缓冲区,这样我就不必每次都去写文件了.这是我迷路的地方.加上创建目录似乎也消耗了太多时间.

我搜索CPAN,网络找到一些可以点亮我的路线但没有运气的线索.有人在这个问题上有建议吗？我应该在哪里阅读或如何进行？我相信Perl能够解决这个问题,但我想我没有使用正确的工具.

open(my $data,"<", $file);
my @lines = <$data>;

foreach (@lines) {
    chomp $_;
    my $line = $_;

    my @each = split(' ',$line);
    if (@each == 10) {
       my @logt = split('/',$each[3]);
       my $llg=1;

       if ($logt[1] == "200") {
           $llg = 9;
       }

       my $urln = new URI::URL $each[6];
       my $netl = $urln->netloc;

       my $flnm = md5_hex($netl);
       my $urlm = md5_hex($each[6]);

       if ( ! -d $outp."/".$flnm ) {
          mkdir $outp."/".$flnm,0644;
       }

       open(my $csvf,">>".$outp."/".$flnm."/".$time."_".$urlm) or die $!;
       print $csvf int($each[0]).";".$each[2].";".$llg."\n";
       close $csvf;   #--->> I want to get rid of this so I can use buffer      
    }
    else {
       print $badf $line;
    }

}

假设上面的代码在子例程中使用并且被线程化了12次.上面代码的参数是文件名.我想摆脱关闭.每次打开和关闭文件时都会调用系统I/O,导致速度变慢.这是我的假设当然,我更愿意接受任何建议

提前致谢

Answer 1

您可能会多次打开同一个文件.如果是这样,那么在数据结构中收集信息可能是有益的,并且仅在循环完成后写入文件.这样可以避免重复测试同一目录的存在,并且只打开每个输出文件一次.

我们也应该摆脱URI::URL- 考虑到你的性能要求,在每次循环迭代期间创建一个新对象太昂贵了.如果您的网址看起来像http://user:password@example.com/path/或https://example.com/,我们可以使用简单的正则表达式.

open my $data, "<", $file or die "Can't open $file: $!";

my %entries;  # collect entries here during the loop

# only read one line at a time, don't keep unnecessary ballast around
while (my $line = <$data>) {
    chomp $line;

    my @each = split(' ',$line);

    if (@each != 10) {
        print $badf $line;
        next;
    }

    my (undef, $logt) = split('/', $each[3]);
    my $llg = ($logt == 200) ? 9 : 1;

    my $url = $each[6];
    my ($server) = $url =~ m{\A\w+://([^/]+)};

    push @{ $entries{$server}{$url} }, sprintf "%d;%s;%d\n", $each[0], $each[2], $llg;
}

while (my ($dir, $files) = each %entries) {
    my $dir_hash = md5_hex($dir);
    my $dirname = "$outp/$dir_hash";

    mkdir $dirname, 0644 or die "Can't create $dirname: $!" unless -d $dirname;

    while (my ($file, $lines) = each %$files) {
        my $file_hash = md5_hex($file);
        my $filename = "$dirname/${time}_${file_hash}";

        open my $csv_fh, ">>", $filename or die "Can't open $filename: $!";
        print { $csv_fh } @$lines;
    }
}

我还清理了代码的其他方面(例如变量命名,错误处理).我将调用md5_hex移出主循环,但根据数据的类型,最好不要延迟散列.