加快内核估计的采样速度



这是我正在使用的一个更大代码的MWE。基本上,它对位于某个阈值以下的所有值的KDE(内核密度估计)执行蒙特卡罗积分(在这个问题上建议了积分方法BTW:积分2D内核密度估计值)。

import numpy as np
from scipy import stats
import time
# Generate some random two-dimensional data:
def measure(n):
    "Measurement model, return two coupled measurements."
    m1 = np.random.normal(size=n)
    m2 = np.random.normal(scale=0.5, size=n)
    return m1+m2, m1-m2
# Get data.
m1, m2 = measure(20000)
# Define limits.
xmin = m1.min()
xmax = m1.max()
ymin = m2.min()
ymax = m2.max()
# Perform a kernel density estimate on the data.
x, y = np.mgrid[xmin:xmax:100j, ymin:ymax:100j]
values = np.vstack([m1, m2])
kernel = stats.gaussian_kde(values)
# Define point below which to integrate the kernel.
x1, y1 = 0.5, 0.5
# Get kernel value for this point.
tik = time.time()
iso = kernel((x1,y1))
print 'iso: ', time.time()-tik
# Sample from KDE distribution (Monte Carlo process).
tik = time.time()
sample = kernel.resample(size=1000)
print 'resample: ', time.time()-tik
# Filter the sample leaving only values for which
# the kernel evaluates to less than what it does for
# the (x1, y1) point defined above.
tik = time.time()
insample = kernel(sample) < iso
print 'filter/sample: ', time.time()-tik
# Integrate for all values below iso.
tik = time.time()
integral = insample.sum() / float(insample.shape[0])
print 'integral: ', time.time()-tik

输出看起来像这样:

iso:  0.00259208679199
resample:  0.000817060470581
filter/sample:  2.10829401016
integral:  4.2200088501e-05

这显然意味着filter/sample调用几乎占用了代码运行所需的所有时间。我必须反复运行这个代码块几千次,这样它可能会非常耗时。

有什么方法可以加快过滤/采样过程吗?


添加

以下是我实际代码的一个稍微现实一点的MWE,其中包含Ophion的多线程解决方案:

import numpy as np
from scipy import stats
from multiprocessing import Pool
def kde_integration(m_list):
    m1, m2 = [], []
    for item in m_list:
        # Color data.
        m1.append(item[0])
        # Magnitude data.
        m2.append(item[1])
    # Define limits.
    xmin, xmax = min(m1), max(m1)
    ymin, ymax = min(m2), max(m2)
    # Perform a kernel density estimate on the data:
    x, y = np.mgrid[xmin:xmax:100j, ymin:ymax:100j]
    values = np.vstack([m1, m2])
    kernel = stats.gaussian_kde(values)
    out_list = []
    for point in m_list:
        # Compute the point below which to integrate.
        iso = kernel((point[0], point[1]))
        # Sample KDE distribution
        sample = kernel.resample(size=1000)
        #Create definition.
        def calc_kernel(samp):
            return kernel(samp)
        #Choose number of cores and split input array.
        cores = 4
        torun = np.array_split(sample, cores, axis=1)
        #Calculate
        pool = Pool(processes=cores)
        results = pool.map(calc_kernel, torun)
        #Reintegrate and calculate results
        insample_mp = np.concatenate(results) < iso
        # Integrate for all values below iso.
        integral = insample_mp.sum() / float(insample_mp.shape[0])
        out_list.append(integral)
    return out_list

# Generate some random two-dimensional data:
def measure(n):
    "Measurement model, return two coupled measurements."
    m1 = np.random.normal(size=n)
    m2 = np.random.normal(scale=0.5, size=n)
    return m1+m2, m1-m2
# Create list to pass.
m_list = []
for i in range(60):
    m1, m2 = measure(5)
    m_list.append(m1.tolist())
    m_list.append(m2.tolist())
# Call KDE integration function.
print 'Integral result: ', kde_integration(m_list)

Ophion提出的解决方案在我提出的原始代码上运行良好,但在此版本中失败,出现以下错误:

Integral result: Exception in thread Thread-3:
Traceback (most recent call last):
  File "/usr/lib/python2.7/threading.py", line 551, in __bootstrap_inner
    self.run()
  File "/usr/lib/python2.7/threading.py", line 504, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 319, in _handle_tasks
    put(task)
PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

我试着移动calc_kernel函数,因为这个问题的答案之一是多处理:如何在类中定义的函数上使用Pool.map?声明"您提供给map()的函数必须可以通过导入您的模块来访问";但我仍然无法使此代码正常工作。

任何帮助都将不胜感激。


添加2

执行Ophion的建议删除calc_kernel函数,并简单地使用:

results = pool.map(kernel, torun)

可以去掉PicklingError,但现在我发现,如果我创建一个超过62-63个项目的初始m_list,我会得到这个错误:

Traceback (most recent call last):
  File "~/gauss_kde_temp.py", line 67, in <module>
    print 'Integral result: ', kde_integration(m_list)
  File "~/gauss_kde_temp.py", line 38, in kde_integration
    pool = Pool(processes=cores)
  File "/usr/lib/python2.7/multiprocessing/__init__.py", line 232, in Pool
    return Pool(processes, initializer, initargs, maxtasksperchild)
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 161, in __init__
    self._result_handler.start()
  File "/usr/lib/python2.7/threading.py", line 494, in start
    _start_new_thread(self.__bootstrap, ())
thread.error: can't start new thread

由于我在这个代码的实际实现中的实际列表最多可以有2000个项目,所以这个问题导致代码无法使用。38行是这样的:

pool = Pool(processes=cores)

所以很明显,这与我使用的内核数量有关?

这个问题";可以';t启动一个新的线程错误"t;在Python中建议使用:

threading.active_count()

当我收到错误时,检查我的线程数。我检查了一下,当它到达374线程时,它总是崩溃。如何围绕此问题编写代码?


这是处理最后一个问题的新问题:线程错误:can';t启动新线程

可能加快速度的最简单方法是并行化kernel(sample):

取此代码片段:

tik = time.time()
insample = kernel(sample) < iso
print 'filter/sample: ', time.time()-tik
#filter/sample:  1.94065904617

将其更改为使用multiprocessing:

from multiprocessing import Pool
tik = time.time()
#Create definition.
def calc_kernel(samp):
    return kernel(samp)
#Choose number of cores and split input array.
cores = 4
torun = np.array_split(sample, cores, axis=1)
#Calculate
pool = Pool(processes=cores)
results = pool.map(calc_kernel, torun)
#Reintegrate and calculate results
insample_mp = np.concatenate(results) < iso
print 'multiprocessing filter/sample: ', time.time()-tik
#multiprocessing filter/sample:  0.496874094009

仔细检查他们是否返回相同的答案:

print np.all(insample==insample_mp)
#True

在4个核心上提高了3.9倍。不确定你在什么上运行这个,但在大约6个处理器之后,你的输入数组大小不够大,无法获得显著的增益。例如,使用20个处理器,其速度仅为5.8倍。

本文评论部分(下面的链接)中的声明是

SciPy的gaussian_kde不使用FFT,而有一个统计模型实现可以使用

…这可能是观察到的性能不佳的原因。它接着报告了使用FFT的数量级改进。请参阅@jseabold的回复。

http://slendrmeans.wordpress.com/2012/05/01/will-it-python-machine-learning-for-hackers-chapter-2-part-1-summary-stats-and-density-estimators/

免责声明:我没有统计模型或科幻的经验。

最新更新