如何有效地计算两个字典的内积答案

【问题标题】：How to efficiently compute the inner product of two dictionaries如何有效地计算两个字典的内积
【发布时间】：2013-08-16 08:03:33
【问题描述】：

假设我使用字典表示一个特征向量（为什么？因为我知道这些特征是稀疏的，但是稍后会详细介绍）。

我应该如何实现两个这样的字典的内积（表示，A，B）

我尝试了幼稚的方法：

for k in A:
  if k in B:
    sum += A[k] * B[k]

但事实证明它很慢。

更多细节：

我使用字典来表示特征，因为
1. 功能键是字符串
2. 有大约 20K 个可能的键
3. 每个向量都是稀疏的（例如，大约 1000 个非零元素）。
我对计算 N=2000 个不同字典（即它们的线性核）的成对内积非常感兴趣。

【问题讨论】：

你可以试试for k, v in A.iteritems(): sum += v * B.get(k,0)
这可能很明显，但我想我还是要添加它：您是否确保“A”具有两个字典中最小的len()？根据具体情况，这可能会对速度产生巨大影响。（并且由于该操作是可交换的，因此不会对答案产生任何影响。）

标签： python linear-algebra

【解决方案1】：

不确定更快，但这是另一种方法：

keys = A.viewkeys() & B.viewkeys()
the_sum = sum(a[k] * b[k] for k in keys)

【讨论】：

您可以在 Python 3 上使用 A.keys() & B.keys()，在 Python 2.7 上使用 A.viewkeys() & B.viewkeys()。
实际上set(A).intersection(B) 在我刚刚做的一些基准测试中看起来更快。

【解决方案2】：

嗯，看来您的方法实际上是最适合密集向量的方法：

>>> # Eric's answer
>>> timeit.timeit('sum([A[k]*B[k] for k in set(A.keys()) & set(B.keys())])', setup='A=dict((i,i) for i in xrange(100));B=dict((i,i) for i in xrange(100))', number=10000)
0.4360210521285808

>>> # My comment
>>> timeit.timeit('for k,v in A.iteritems(): sum += v*B.get(k,0)', setup='A=dict((i,i) for i in xrange(100));B=dict((i,i) for i in xrange(100));sum=0', number=10000)
0.4082838999682963

# My comment, more compact
>>> timeit.timeit('sum(v*B.get(k,0) for k,v in A.iteritems())', setup='A=dict((i,i) for i in xrange(100));B=dict((i,i) for i in xrange(100))', number=10000)
0.38053266868496394

>>> #Your approach
>>> timeit.timeit('for k in A: sum += A[k]*B[k] if k in B else 0.', setup='A=dict((i,i) for i in xrange(100));B=dict((i,i) for i in xrange(100));sum=0', number=10000)
0.35574231962510794

>>> # Your approach, more compact
>>> timeit.timeit('sum(A[k]*B[k] for k in A if k in B)', setup='A=dict((i,i) for i in xrange(100));B=dict((i,i) for i in xrange(100))', number=10000)
0.3400850549682559

对于稀疏的，埃里克的答案表现更好，但你的仍然是最快的：

# Mine
>>> timeit.timeit('sum(v*B.get(k,0) for k,v in A.iteritems())', setup='import random;A=dict((i,i) for i in xrange(100) if random.random() < 0.3);B=dict((i,i) for i in xrange(100) if random.random() < 0.2)', number=10000)
0.1390782696843189

# Eric's
>>> timeit.timeit('sum([A[k]*B[k] for k in set(A.keys()) & set(B.keys())])', setup='import random;A=dict((i,i) for i in xrange(100) if random.random() < 0.3);B=dict((i,i) for i in xrange(100) if random.random() < 0.2)', number=10000)
0.11702822992151596

# Yours
>>> timeit.timeit('sum(A[k]*B[k] for k in A if k in B)', setup='import random;A=dict((i,i) for i in xrange(100) if random.random() < 0.3);B=dict((i,i) for i in xrange(100) if random.random() < 0.2)', number=10000)
0.07878250570843193

编辑

经过一番折腾，sum([x for x ...]) 似乎比sum(x for x in ...) 快得多。用这个和 Janne 对 Eric 答案中关键点的评论进行重新基准测试，你的仍然是最重要的（Joowani 略有改进）：

>>> timeit.timeit('sum([v*B.get(k,0) for k,v in A.items()])', setup='import random;A=dict((i,i) for i in xrange(100) if random.random() < 0.3);B=dict((i,i) for i in xrange(100) if random.random() < 0.2)', number=100000)
1.1604375791416714
>>> timeit.timeit('sum([A[k]*B[k] for k in A.viewkeys() & B.viewkeys()])', setup='import random;A=dict((i,i) for i in xrange(100) if random.random() < 0.3);B=dict((i,i) for i in xrange(100) if random.random() < 0.2)', number=100000)
0.9234189571552633
>>> timeit.timeit('sum([A[k]*B[k] for k in A if k in B])', setup='import random;A=dict((i,i) for i in xrange(100) if random.random() < 0.3);B=dict((i,i) for i in xrange(100) if random.random() < 0.2)', number=100000)
0.5411289579401455
>>> timeit.timeit('sum([A[k]*B[k] for k in A if k in B]) if len(A)<len(B) else sum([A[k]*B[k] for k in B if k in A])', setup='import random;A=dict((i,i) for i in xrange(100) if random.random() < 0.3);B=dict((i,i) for i in xrange(100) if random.random() < 0.2)', number=100000)
0.5198972138696263

缩放到非常大的尺寸，您会看到完全相同的模式：

>>> #Mine
>>> timeit.timeit('sum([v*B.get(k,0) for k,v in A.iteritems()])', setup='import random;A=dict((i,i) for i in xrange(10000) if random.random() < 0.1);B=dict((i,i) for i in xrange(10000) if random.random() < 0.2)', number=100000)
45.328807250833506

>>> #Eric's
>>> timeit.timeit('sum([A[k]*B[k] for k in A.viewkeys() & B.viewkeys()])', setup='import random;A=dict((i,i) for i in xrange(10000) if random.random() < 0.1);B=dict((i,i) for i in xrange(10000) if random.random() < 0.2)', number=100000)
28.042937058640973

>>> #Yours
>>> timeit.timeit('sum([A[k]*B[k] for k in A if k in B])', setup='import random;A=dict((i,i) for i in xrange(10000) if random.random() < 0.1);B=dict((i,i) for i in xrange(10000) if random.random() < 0.2)', number=100000)
16.55080344861699

>>> #Joowani's
>>> timeit.timeit('sum([A[k]*B[k] for k in A if k in B]) if len(A)<len(B) else sum([A[k]*B[k] for k in B if k in A])', setup='import random;A=dict((i,i) for i in xrange(10000) if random.random() < 0.1);B=dict((i,i) for i in xrange(10000) if random.random() < 0.2)', number=100000)
15.485236119691308

我认为 Joowani 的技巧在这里并没有显着改善它，因为向量的大小大致相同，但根据您的问题（如果某些向量比其他向量小得离谱），这可能更重要...

再次编辑

哎呀，好像我应该在发帖之前再喝杯咖啡...正如 Eric 指出的（尽管我完全错过了...），在 setup 中定义数组使其在所有试验中保持相同，即并不是真正的基准测试的最佳方法。使用 PROPER 随机向量进行测试，结果没有显着差异，但为了完整起见：

>>> timeit.timeit('mine(dict((i,i) for i in xrange(100) if random.random() < 0.3),dict((i,i) for i in xrange(100) if random.random() < 0.2))', setup='import random;joowanis=lambda A,B:sum([A[k]*B[k] for k in A if k in B]) if len(A)<len(B) else sum([A[k]*B[k] for k in B if k in A]);mine=lambda A,B:sum([v*B.get(k,0) for k,v in A.iteritems()]);erics=lambda A,B:sum([A[k]*B[k] for k in A.viewkeys() & B.viewkeys()]);yours=lambda A,B:sum([A[k]*B[k] for k in A if k in B])', number=100000)
6.294158102577967
>>> timeit.timeit('erics(dict((i,i) for i in xrange(100) if random.random() < 0.3),dict((i,i) for i in xrange(100) if random.random() < 0.2))', setup='import random;joowanis=lambda A,B:sum([A[k]*B[k] for k in A if k in B]) if len(A)<len(B) else sum([A[k]*B[k] for k in B if k in A]);mine=lambda A,B:sum([v*B.get(k,0) for k,v in A.iteritems()]);erics=lambda A,B:sum([A[k]*B[k] for k in A.viewkeys() & B.viewkeys()]);yours=lambda A,B:sum([A[k]*B[k] for k in A if k in B])', number=100000)
6.068052507449011
>>> timeit.timeit('yours(dict((i,i) for i in xrange(100) if random.random() < 0.3),dict((i,i) for i in xrange(100) if random.random() < 0.2))', setup='import random;joowanis=lambda A,B:sum([A[k]*B[k] for k in A if k in B]) if len(A)<len(B) else sum([A[k]*B[k] for k in B if k in A]);mine=lambda A,B:sum([v*B.get(k,0) for k,v in A.iteritems()]);erics=lambda A,B:sum([A[k]*B[k] for k in A.viewkeys() & B.viewkeys()]);yours=lambda A,B:sum([A[k]*B[k] for k in A if k in B])', number=100000)
5.745110704570834
>>> timeit.timeit('joowanis(dict((i,i) for i in xrange(100) if random.random() < 0.3),dict((i,i) for i in xrange(100) if random.random() < 0.2))', setup='import random;joowanis=lambda A,B:sum([A[k]*B[k] for k in A if k in B]) if len(A)<len(B) else sum([A[k]*B[k] for k in B if k in A]);mine=lambda A,B:sum([v*B.get(k,0) for k,v in A.iteritems()]);erics=lambda A,B:sum([A[k]*B[k] for k in A.viewkeys() & B.viewkeys()]);yours=lambda A,B:sum([A[k]*B[k] for k in A if k in B])', number=100000)
5.737499445367575

缩放：

>>> timeit.timeit('mine(dict((i,i) for i in xrange(10000) if random.random() < 0.1),dict((i,i) for i in xrange(10000) if random.random() < 0.2))', setup='import random;joowanis=lambda A,B:sum([A[k]*B[k] for k in A if k in B]) if len(A)<len(B) else sum([A[k]*B[k] for k in B if k in A]);mine=lambda A,B:sum([v*B.get(k,0) for k,v in A.iteritems()]);erics=lambda A,B:sum([A[k]*B[k] for k in A.viewkeys() & B.viewkeys()]);yours=lambda A,B:sum([A[k]*B[k] for k in A if k in B])', number=1000)
5.0510995368395015
>>> timeit.timeit('erics(dict((i,i) for i in xrange(10000) if random.random() < 0.1),dict((i,i) for i in xrange(10000) if random.random() < 0.2))', setup='import random;joowanis=lambda A,B:sum([A[k]*B[k] for k in A if k in B]) if len(A)<len(B) else sum([A[k]*B[k] for k in B if k in A]);mine=lambda A,B:sum([v*B.get(k,0) for k,v in A.iteritems()]);erics=lambda A,B:sum([A[k]*B[k] for k in A.viewkeys() & B.viewkeys()]);yours=lambda A,B:sum([A[k]*B[k] for k in A if k in B])', number=1000)
4.350612399185138
>>> timeit.timeit('yours(dict((i,i) for i in xrange(10000) if random.random() < 0.1),dict((i,i) for i in xrange(10000) if random.random() < 0.2))', setup='import random;joowanis=lambda A,B:sum([A[k]*B[k] for k in A if k in B]) if len(A)<len(B) else sum([A[k]*B[k] for k in B if k in A]);mine=lambda A,B:sum([v*B.get(k,0) for k,v in A.iteritems()]);erics=lambda A,B:sum([A[k]*B[k] for k in A.viewkeys() & B.viewkeys()]);yours=lambda A,B:sum([A[k]*B[k] for k in A if k in B])', number=1000)
4.15619379016789
>>> timeit.timeit('joowanis(dict((i,i) for i in xrange(10000) if random.random() < 0.1),dict((i,i) for i in xrange(10000) if random.random() < 0.2))', setup='import random;joowanis=lambda A,B:sum([A[k]*B[k] for k in A if k in B]) if len(A)<len(B) else sum([A[k]*B[k] for k in B if k in A]);mine=lambda A,B:sum([v*B.get(k,0) for k,v in A.iteritems()]);erics=lambda A,B:sum([A[k]*B[k] for k in A.viewkeys() & B.viewkeys()]);yours=lambda A,B:sum([A[k]*B[k] for k in A if k in B])', number=1000)
4.185129374341159

我认为底线是，你不能指望通过巧妙地重新排序这种事情的表达式来显着加速......也许你可以尝试在 C/Cython 中执行数字部分或使用 Scipy's Sparse 包？

【讨论】：

我刚刚将我的从 set(d.keys()) 更新为 d.viewkeys()，这可能会提高速度
另外，如果每个代码 sn-p 的初始字典不同，我不确定这个测试是否有意义
没错，但我认为这仍然可以让我们大致了解每个人的平均速度。
看来我的方式确实加快了我的速度。我可以请您确认一下吗？
恭喜！你来了 :)

【解决方案3】：

如果 A 比 B 长得多，这可能会有所帮助吗？

if len(A) > len(B):
    A, B = B, A

for k in A:
    if k in B:
        the_sum += A[k] * B[k]

【讨论】：

【解决方案4】：

你应该尝试使用 namedtuples 而不是 dict。

from collections import namedtuple
A = dict
B = dict
_A = namedtuple('_A', A.keys())
_B = namedtuple('_B', B.keys())
DictA = _A(**A)
DictB = _B(**B)

然后将它们用作字典。有关命名元组的更多信息：What are "named tuples" in Python?

【讨论】：

这将如何加快计算速度？
其实namedtuples比dict(stackoverflow.com/questions/1336791/…)更快更高效，用that代替dict可以提高计算速度
如果您按名称查找字段，它仍然是引擎盖下的字典查找。

【解决方案5】：

这是我的答案（根据@valentin-clement 的建议）：

首先我包装了一个 scipy.sparse dok_matrix。这个想法是为每个可能的特征分配一个索引。

import scipy.sparse as sps
import numpy as np

class MSK:
    # DD is a dict of dict, whose values are of type float.
    # features - the set of possible features keys
    def __init__(self, DD, features):
        self.features = {k: j for (j, k) in enumerate(features)}
        self.strings = DD.keys()
        n = len(self.strings)
        d = len(self.features)
        self.M = sps.dok_matrix((n, d), dtype=np.float64)
        for (i, s) in enumerate(self.strings):
            v = DD[s]
            for k in v:
                j = self.features[k]
                self.M[i, j] = v[k]

并且我们使用下面的代码进行测试，其中元素个数是800，维度也是800，但是稀疏度是200（正好200个元素不为零）。

np.random.seed(1)
N = 800
DD = dict()
R = range(N)
for i in xrange(N):
    DD[i] = dict()
    S = np.random.permutation(R)
    S = S[:N/4]
    for j in S:
        DD[i][j] = np.random.randn(1)[0]

K = MSK(DD, R)
import cProfile
cProfile.runctx("A = K.M * K.M.T", globals(), locals())
print A.todense()

输出是：

2080520 function calls (2080519 primitive calls) in 2.884 seconds

假设 3 秒。我的幼稚实现（使用@Joowani 的 if 语句）大约需要 19 秒。

（MSK 代表 MatrixSparseKeys）

【讨论】：

仅供参考，将 dok_matrix() 替换为 lil_matrix() 我们可以获得更好的性能。