1

我有一个大约 50,000 个用户的数组参考。我想遍历所有这些用户并将每个用户与所有其他用户进行比较，以建立一个加权匹配列表（如果名称完全匹配，则值得 x，部分匹配值得 y 等）。

在浏览完列表并进行所有检查之后，我想要获得 10 个权重最高的匹配项。这是我正在做的一个例子来帮助解释：

#!/usr/bin/perl
######################################################################
# Libraries
# ---------
use strict;
use warnings;

my $users = [];
$users->[0]{'Name'} = 'xxx';
$users->[0]{'Address'} = 'yyyy';
$users->[0]{'Phone'} = 'xxx';
$users->[1]{'Name'} = 'xxx';
$users->[1]{'Address'} = 'yyyy';
$users->[1]{'Phone'} = 'xxx';
$users->[2]{'Name'} = 'xxx';
$users->[3]{'Address'} = 'yyyy';
$users->[4]{'Phone'} = 'xxx';
foreach my $user_to_check (@$users) {
    my $matched_users = [];
    foreach my $user (@$users) {            
        $user_to_check->{'Weight'} = 0;
        if (lc($user_to_check->{'Name'}) eq lc($user->{'Name'})) {
            $user_to_check->{'Weight'} = ($user_to_check->{'Weight'} + 10);
        } elsif ((length($user_to_check->{'Name'}) > 2) && (length($user->{'Name'}) > 2) && ($user_to_check->{'Name'} =~ /\Q$user->{'Name'}\E/i)) {
            $user_to_check->{'Weight'} = ($user_to_check->{'Weight'} + 5);
        }
        if (lc($user_to_check->{'Address'}) eq lc($user->{'Address'})) {
            .....
        }
        if ($user_to_check->{'Weight'} > 0) {
            # We have matches, add to matched users
            push (@$matched_users,$user);
        }
    }
   # Now we want to get just the top 10 highest matching users
   foreach my $m_user (sort { $b->{'Weight'} <=> $a->{'Weight'} } @$matched_users ) {
    last if $counter == 10;
       .... # Do stuff with the 10 we want
    }         
}

问题是，太慢了。运行需要一天多的时间（我已经在多台机器上尝试过）。我知道“排序”是一个杀手，但我也尝试将结果插入到 tmp mysql 表中，然后在最后而不是执行 Perl 排序，我只是按选择进行排序，但时间差异非常次要的。

由于我只是在浏览现有的数据结构并进行比较，我不确定我能做些什么（如果有的话）来加速它。我会很感激任何建议。

4

1 回答 1

15

O(n²)

您将其中的每个元素@$users与其中的每个元素进行比较。即 5E4² = 2.5E9 比较。例如，您不需要将元素与其自身进行比较。您也不需要将一个元素与您已经比较过的元素进行比较。即在这个比较表中

  X Y Z
X - + +
Y - - +
Z - - -

只需进行三个比较即可将每个元素与所有其他元素进行比较。你正在做的九个比较有 66% 是不必要的（渐近：50% 是不必要的）。

您可以通过循环索引来实现这一点：

for my $i (0 .. $#$users) {
  my $userA = $users->[$i];
  for my $j ($i+1 .. $#$users) {
    my $userB = $users->[$j];
    ...;
  }
}

但这意味着在匹配时，您必须增加两个匹配用户的权重。

做一件事，而不是 100,000 次

您将每个用户的名称小写 1E5 次。这是 1E5 - 1 倍！只需对每个元素执行一次，可能在数据输入时。

作为旁注，您不应该执行小写，您应该进行大小写折叠。从至少 v16 开始，通过该功能可以使用此fc功能。当您有非英文数据时，仅小写将是错误的。

use feature 'fc'; # needs v16
$user->[NAME] = fc $name;

或者

use Unicode::CaseFold;
$user->[NAME] = fc $name;

当哈希不够快时

哈希很快，因为查找需要恒定的时间。但是单个哈希查找比数组访问更昂贵。由于您只有一小部分预定义的字段，您可以使用以下技巧来使用类似哈希的数组：

用映射到索引的字段名称声明一些常量，例如

use constant {
  WEIGHT => 0,
  NAME => 1,
  ADDRESS => 2,
  ...;
};

然后将您的数据放入数组中：

$users->[0][NAME] = $name; ...;

您可以访问以下字段

$userA->[WEIGHT] += 10;

虽然这看起来像一个哈希，但这实际上是一种安全的方法，可以以最小的开销仅访问数组的某些字段。

正则表达式很慢

好吧，它们很快，但是有一个更好的方法来确定一个字符串是否是另一个字符串的子字符串：使用index. IE

$user_to_check->{'Name'} =~ /\Q$user->{'Name'}\E/i

可以写成

(-1 != index $user_to_check->{Name}, $user->{Name})

假设两者都已经小写案例折叠。

替代实施

编辑：这似乎因您对问题的编辑而无效。这假设您试图找到一些全局相似性，而不是为每个用户获得一组好的匹配

实现这些想法会让你的循环看起来有点像

for my $i (0 .. $#$users) {
  my $userA = $users->[$i];
  for my $j ($i+1 .. $#$users) {
    my $userB = $users->[$j];
    if ($userA->[NAME] eq $userB->[NAME]) {
        $userA->[WEIGHT] += 10;
        $userB->[WEIGHT] += 10;
    } elsif ((length($userA->[NAME]) > 2) && (length($userB->[NAME]) > 2))
        $userA->[WEIGHT] += 5 if -1 != index $userA->[NAME], $userB->[NAME];
        $userB->[WEIGHT] += 5 if -1 != index $userB->[NAME], $userA->[NAME];
    }
    if ($userA->[ADDRESS] eq $userB->[ADDRESS]) {
        ..... # More checks
    }
  }
}
my (@top_ten) = (sort { $b->[WEIGHT] <=> $a->[WEIGHT] } @$users)[0 .. 9];

分而治之

您展示的任务是高度可并行化的。如果你有记忆，在这里使用线程很容易：

my $top10 = Thread::Queue->new;
my $users = ...; # each thread gets a copy of this data

my @threads = map threads->create(\&worker, $_), [0, int($#$users/2)], [int($#$users/2)+1, $#users];

# process output from the threads
while (defined(my $ret = $top10->dequeue)) {
  my ($user, @top10) = @$ret;
  ...;
}

$_->join for @threads;

sub worker {
  my ($from, $to) = @_;
  for my $i ($from .. $to) {
    my $userA = $users->[$i];
    for $userB (@$users) {
      ...;
    }
    my @top10 = ...;
    $top10->enqueue([ $userA, @top10 ]); # yield data to the main thread
  }
}

您可能应该通过队列返回您的输出（如此处所示），但在线程内进行尽可能多的处理。通过更高级的工作负载分区，应该产生与可用处理器一样多的线程。

但是，如果任何类型的流水线、过滤或缓存可以减少嵌套循环中所需的迭代次数，那么您应该进行此类优化（想想 map-reduce 风格的编程）。

编辑：通过重复数据删除的哈希优雅地降低复杂性

我们本质上在做的是计算我们的记录匹配程度的矩阵，例如

  X Y Z
X 9 4 5
Y 3 9 2
Z 5 2 9

如果我们假设X 与 Y相似意味着Y 与 X 相似，那么矩阵是对称的，我们只需要它的一半：

  X Y Z
X \ 4 5
Y   \ 2
Z     \

这样的矩阵等价于加权无向图：

4  X  5   |  X – Y: 4
  / \     |  X – Z: 5
 Y---Z    |  Y – Z: 2
   2      |

因此，我们可以将其优雅地表示为散列的散列：

my %graph;
$graph{X}{Y} = 4;
$graph{X}{Z} = 5;
$graph{Y}{Z} = 2;

然而，这样的哈希结构意味着一个方向（从节点X到节点Y）。为了让查询数据更容易，我们不妨也包括另一个方向（由于哈希的实现，这不会导致大量的内存增加）。

$graph{$x}{$y} = $graph{$y}{$x} += 2;

因为每个节点现在只连接到与其相似的那些节点，所以我们不必对 50,000 条记录进行排序。对于第 100 条记录，我们可以获得十个最相似的节点，例如

my $node = 100;
my @top10 = (sort { $graph{$node}{$b} <=> $graph{$node}{$a} } keys %{ $graph{$node} })[0 .. 9];

这会将实现更改为

my %graph;

# build the graph, using the array indices as node ID
for my $i (0 .. $#$users) {
  my $userA = $users->[$i];
  for my $j ($i+1 .. $#$users) {
    my $userB = $users->[$j];
    if ($userA->[NAME] eq $userB->[NAME]) {
        $graph{$j}{$i} = $graph{$i}{$j} += 10;
    } elsif ((length($userA->[NAME]) > 2) && (length($userB->[NAME]) > 2))
        $graph{$j}{$i} = $graph{$i}{$j} += 5
          if -1 != index $userA->[NAME], $userB->[NAME]
          or -1 != index $userB->[NAME], $userA->[NAME];
    }
    if ($userA->[ADDRESS] eq $userB->[ADDRESS]) {
        ..... # More checks
    }
  }
}

# the graph is now fully populated.

# do somethething with each top10
while (my ($node_id, $similar) = each %graph) {
  my @most_similar_ids = (sort { $similar->{$b} <=> $similar->{$a} } keys %$similar)[0 .. 9];
  my ($user, @top10) = @$users[ $node_id, @most_similar_ids ];
  ...;
}

以这种方式构建图应该花费幼稚迭代的一半时间，并且如果每个节点的平均边数足够低，则通过相似节点应该会更快。

并行化这有点困难，因为每个线程生成的图形必须在查询数据之前组合起来。为此，最好每个线程都执行上述代码，但迭代边界作为参数给出，并且只产生一条边。这对边将在组合阶段完成：

THREAD A [0 .. 2/3]   partial
                   \  graph
                    =====> COMBINE -> full graph -> QUERY
                   /  partial
THREAD B [2/3 .. 1]   graph

# note bounds recognizing the triangular distribution of workload

但是，这仅在给定节点只有非常少的相似节点时才有用，因为组合成本很高。

于 2013-07-07T11:58:16.063 回答