我对如何尽可能快地以numpy计算距离有疑问,
def getR1(VVm,VVs,HHm,HHs): t0=time.time() R=VVs.flatten()[numpy.newaxis,:]-VVm.flatten()[:,numpy.newaxis] R*=R R1=HHs.flatten()[numpy.newaxis,:]-HHm.flatten()[:,numpy.newaxis] R1*=R1 R+=R1 del R1 print "R1\t",time.time()-t0, R.shape, #11.7576191425 (108225, 10500) print numpy.max(R) #4176.26290975 # uses 17.5Gb ram return R def getR2(VVm,VVs,HHm,HHs): t0=time.time() precomputed_flat = numpy.column_stack((VVs.flatten(), HHs.flatten())) measured_flat = numpy.column_stack((VVm.flatten(), HHm.flatten())) deltas = precomputed_flat[None,:,:] - measured_flat[:, None, :] #print time.time()-t0, deltas.shape # 5.861109972 (108225, 10500, 2) R = numpy.einsum('ijk,ijk->ij', deltas, deltas) print "R2\t",time.time()-t0,R.shape, #14.5291359425 (108225, 10500) print numpy.max(R) #4176.26290975 # uses 26Gb ram return R def getR3(VVm,VVs,HHm,HHs): from numpy.core.umath_tests import inner1d t0=time.time() precomputed_flat = numpy.column_stack((VVs.flatten(), HHs.flatten())) measured_flat = numpy.column_stack((VVm.flatten(), HHm.flatten())) deltas = precomputed_flat[None,:,:] - measured_flat[:, None, :] #print time.time()-t0, deltas.shape # 5.861109972 (108225, 10500, 2) R = inner1d(deltas, deltas) print "R3\t",time.time()-t0, R.shape, #12.6972110271 (108225, 10500) print numpy.max(R) #4176.26290975 #Uses 26Gb return R def getR4(VVm,VVs,HHm,HHs): from scipy.spatial.distance import cdist t0=time.time() precomputed_flat = numpy.column_stack((VVs.flatten(), HHs.flatten())) measured_flat = numpy.column_stack((VVm.flatten(), HHm.flatten())) R=spdist.cdist(precomputed_flat,measured_flat, 'sqeuclidean') #.T print "R4\t",time.time()-t0, R.shape, #17.7022118568 (108225, 10500) print numpy.max(R) #4176.26290975 # uses 9 Gb ram return R def getR5(VVm,VVs,HHm,HHs): from scipy.spatial.distance import cdist t0=time.time() precomputed_flat = numpy.column_stack((VVs.flatten(), HHs.flatten())) measured_flat = numpy.column_stack((VVm.flatten(), HHm.flatten())) R=spdist.cdist(precomputed_flat,measured_flat, 'euclidean') #.T print "R5\t",time.time()-t0, R.shape, #15.6070930958 (108225, 10500) print numpy.max(R) #64.6240118667 # uses only 9 Gb ram return R def getR6(VVm,VVs,HHm,HHs): from scipy.weave import blitz t0=time.time() R=VVs.flatten()[numpy.newaxis,:]-VVm.flatten()[:,numpy.newaxis] blitz("R=R*R") # R*=R R1=HHs.flatten()[numpy.newaxis,:]-HHm.flatten()[:,numpy.newaxis] blitz("R1=R1*R1") # R1*=R1 blitz("R=R+R1") # R+=R1 del R1 print "R6\t",time.time()-t0, R.shape, #11.7576191425 (108225, 10500) print numpy.max(R) #4176.26290975 return R
结果在以下时间:
R1 11.7737319469 (108225, 10500) 4909.66881791 R2 15.1279799938 (108225, 10500) 4909.66881791 R3 12.7408981323 (108225, 10500) 4909.66881791 R4 17.3336868286 (10500, 108225) 4909.66881791 R5 15.7530870438 (10500, 108225) 70.0690289494 R6 11.670968771 (108225, 10500) 4909.66881791
虽然最后一个给出的是sqrt((VVm-VVs)^ 2 +(HHm-HHs)^ 2),而其他的给出的是(VVm-VVs)^ 2 +(HHm-HHs)^ 2,但这并不是很重要,因为否则在我的代码中,我将为每个i取R [i ,:]的最小值,而sqrt无论如何都不会影响最小值,(如果我对距离感兴趣,我只需取sqrt(value)即可)在整个阵列上执行sqrt的操作,因此实际上没有时序差异。
问题仍然存在:第一个解决方案为什么是最好的(第二个和第三个解决方案速度较慢的原因是因为deltas = …需要5.8秒,(这也是这两种方法都需要26Gb的原因)),为什么欧几里得比欧几里得慢?
squclidean应该只做(VVm-VVs)^ 2 +(HHm-HHs)^ 2,而我认为它做的事情有所不同。有谁知道如何找到该方法的源代码(C或底部的任何内容)?我认为它确实sqrt((VVm-VVs)^ 2 +(HHm-HHs)^ 2)^ 2(我能想到为什么它会比(VVm-VVs)^ 2 +(HHm-HHs)慢的唯一原因^ 2-我知道这是一个愚蠢的原因,有人有一个更合乎逻辑的理由吗?)
既然我对C一无所知,我该如何用scipy.weave内联?而且该代码是否可以像使用python一样正常编译?还是我需要为此安装特殊的东西?
编辑:好的,我尝试了scipy.weave.blitz,(R6方法),并且速度稍快一些,但是我认为知道C比我还多的人仍然可以提高速度吗?我只是采用了形式为a + = b或* =的行,并查看它们在C中的状态,然后将它们放入blitz语句中,但是我想我是否应该在其中使用带有flatten和newaxis的语句行同样,C也应该更快一些,但是我不知道该怎么做(知道C的人可能会解释吗?)。现在,闪电战的东西和我的第一种方法之间的差异还不够大,无法真正由C vs numpy引起吗?
我想其他类似deltas = …的方法也可以快得多,当我将其放在C中时?
每当您有乘法和和时,请尝试使用点积函数或之一np.einsum。由于要预分配数组,而不是为水平和垂直坐标使用不同的数组,因此将它们堆叠在一起:
np.einsum
precomputed_flat = np.column_stack((svf.flatten(), shf.flatten())) measured_flat = np.column_stack((VVmeasured.flatten(), HHmeasured.flatten())) deltas = precomputed_flat - measured_flat[:, None, :]
从这里开始,最简单的方法是:
dist = np.einsum('ijk,ijk->ij', deltas, deltas)
您也可以尝试以下方法:
from numpy.core.umath_tests import inner1d dist = inner1d(deltas, deltas)
当然也有SciPy的空间模块cdist:
cdist
from scipy.spatial.distance import cdist dist = cdist(precomputed_flat, measured_flat, 'euclidean')
编辑 我无法在如此大的数据集上运行测试,但是这些时间颇具启发性:
len_a, len_b = 10000, 1000 a = np.random.rand(2, len_a) b = np.random.rand(2, len_b) c = np.random.rand(len_a, 2) d = np.random.rand(len_b, 2) In [3]: %timeit a[:, None, :] - b[..., None] 10 loops, best of 3: 76.7 ms per loop In [4]: %timeit c[:, None, :] - d 1 loops, best of 3: 221 ms per loop
对于上述较小的数据集,通过在内存中以不同的方式排列数据,我可以稍微加快使用scipy.spatial.distance.cdist和匹配方法的速度inner1d:
scipy.spatial.distance.cdist
inner1d
precomputed_flat = np.vstack((svf.flatten(), shf.flatten())) measured_flat = np.vstack((VVmeasured.flatten(), HHmeasured.flatten())) deltas = precomputed_flat[:, None, :] - measured_flat import scipy.spatial.distance as spdist from numpy.core.umath_tests import inner1d In [13]: %timeit r0 = a[0, None, :] - b[0, :, None]; r1 = a[1, None, :] - b[1, :, None]; r0 *= r0; r1 *= r1; r0 += r1 10 loops, best of 3: 146 ms per loop In [14]: %timeit deltas = (a[:, None, :] - b[..., None]).T; inner1d(deltas, deltas) 10 loops, best of 3: 145 ms per loop In [15]: %timeit spdist.cdist(a.T, b.T) 10 loops, best of 3: 124 ms per loop In [16]: %timeit deltas = a[:, None, :] - b[..., None]; np.einsum('ijk,ijk->jk', deltas, deltas) 10 loops, best of 3: 163 ms per loop