【问题标题】:Fast replacement of values in a numpy array快速替换 numpy 数组中的值
【发布时间】:2011-03-25 03:44:30
【问题描述】:

我有一个非常大的 numpy 数组(包含多达一百万个元素),如下所示:

[0,1,6,5,1,2,7,6,2,3,8,7,3,4,9,8,5,6,11,10,6,7,12,11,7,
8,13,12,8,9,14,13,10,11,16,15,11,12,17,16,12,13,18,17,13,
14,19,18,15,16,21,20,16,17,22,21,17,18,23,22,18,19,24,23]

以及用于替换上述数组中一些元素的小字典映射

{4: 0, 9: 5, 14: 10, 19: 15, 20: 0, 21: 1, 22: 2, 23: 3, 24: 0}

我想根据上面的地图替换一些元素。 numpy 数组真的很大,只有一小部分元素(作为字典中的键出现)会被相应的值替换。最快的方法是什么?

【问题讨论】:

标签: python replace numpy


【解决方案1】:

我相信还有更有效的方法,但现在,尝试一下

from numpy import copy

newArray = copy(theArray)
for k, v in d.iteritems(): newArray[theArray==k] = v

微基准测试和正确性测试:

#!/usr/bin/env python2.7

from numpy import copy, random, arange

random.seed(0)
data = random.randint(30, size=10**5)

d = {4: 0, 9: 5, 14: 10, 19: 15, 20: 0, 21: 1, 22: 2, 23: 3, 24: 0}
dk = d.keys()
dv = d.values()

def f1(a, d):
    b = copy(a)
    for k, v in d.iteritems():
        b[a==k] = v
    return b

def f2(a, d):
    for i in xrange(len(a)):
        a[i] = d.get(a[i], a[i])
    return a

def f3(a, dk, dv):
    mp = arange(0, max(a)+1)
    mp[dk] = dv
    return mp[a]


a = copy(data)
res = f2(a, d)

assert (f1(data, d) == res).all()
assert (f3(data, dk, dv) == res).all()

结果:

$ python2.7 -m timeit -s 'from w import f1,f3,data,d,dk,dv' 'f1(data,d)'
100 loops, best of 3: 6.15 msec per loop

$ python2.7 -m timeit -s 'from w import f1,f3,data,d,dk,dv' 'f3(data,dk,dv)'
100 loops, best of 3: 19.6 msec per loop

【讨论】:

  • 像for k in d 这样迭代会尽可能快`
  • 投票反对@katrielalex 提到的numpy.place,因为它只是浪费了我大约二十到三十个小时的时间;显然,不鼓励使用它。 “我通常建议使用np.copyto 或(在这种情况下)布尔花式索引来实现相同的目标,并避免使用np.place 或np.putmask。我意识到在某些情况下,这些函数并不是完全1:1 替换的。” FWIW 我没有这个错误,但另一个它只是默默地无法工作。
  • 对于任何想知道的人,命令现在是for k, v in d.items(): newArray[theArray==k] = v
  • 注意:如果 dict 中有 3 倍以上的项目,它会慢 3 倍(我认为 f3 版本不会受到那么多影响)。
【解决方案2】:

假设值介于 0 和某个最大整数之间,可以通过使用 numpy-array 作为int->int dict 来实现快速替换,如下所示

mp = numpy.arange(0,max(data)+1)
mp[replace.keys()] = replace.values()
data = mp[data]

在哪里先

data = [ 0  1  6  5  1  2  7  6  2  3  8  7  3  4  9  8  5  6 11 10  6  7 12 11  7
  8 13 12  8  9 14 13 10 11 16 15 11 12 17 16 12 13 18 17 13 14 19 18 15 16
 21 20 16 17 22 21 17 18 23 22 18 19 24 23]

并替换为

replace = {4: 0, 9: 5, 14: 10, 19: 15, 20: 0, 21: 1, 22: 2, 23: 3, 24: 0}

我们得到

data = [ 0  1  6  5  1  2  7  6  2  3  8  7  3  0  5  8  5  6 11 10  6  7 12 11  7
  8 13 12  8  5 10 13 10 11 16 15 11 12 17 16 12 13 18 17 13 10 15 18 15 16
  1  0 16 17  2  1 17 18  3  2 18 15  0  3]

【讨论】:

【解决方案3】:

实现这一点的另一种更通用的方法是函数向量化:

import numpy as np

data = np.array([0, 1, 6, 5, 1, 2, 7, 6, 2, 3, 8, 7, 3, 4, 9, 8, 5, 6, 11, 10, 6, 7, 12, 11, 7, 8, 13, 12, 8, 9, 14, 13, 10, 11, 16, 15, 11, 12, 17, 16, 12, 13, 18, 17, 13, 14, 19, 18, 15, 16, 21, 20, 16, 17, 22, 21, 17, 18, 23, 22, 18, 19, 24, 23])
mapper_dict = {4: 0, 9: 5, 14: 10, 19: 15, 20: 0, 21: 1, 22: 2, 23: 3, 24: 0}

def mp(entry):
    return mapper_dict[entry] if entry in mapper_dict else entry
mp = np.vectorize(mp)

print mp(data)

【讨论】:

  • 或者只是return mapper_dict.get(entry, entry)
【解决方案4】:

我对一些解决方案进行了基准测试,结果毫无吸引力:

import timeit
import numpy as np

array = 2 * np.round(np.random.uniform(0,10000,300000)).astype(int)
from_values = np.unique(array) # pair values from 0 to 2000
to_values = np.arange(from_values.size) # all values from 0 to 1000
d = dict(zip(from_values, to_values))

def method_for_loop():
    out = array.copy()
    for from_value, to_value in zip(from_values, to_values) :
        out[out == from_value] = to_value
    print('Check method_for_loop :', np.all(out == array/2)) # Just checking
print('Time method_for_loop :', timeit.timeit(method_for_loop, number = 1))

def method_list_comprehension():
    out = [d[i] for i in array]
    print('Check method_list_comprehension :', np.all(out == array/2)) # Just checking
print('Time method_list_comprehension :', timeit.timeit(method_list_comprehension, number = 1))

def method_bruteforce():
    idx = np.nonzero(from_values == array[:,None])[1]
    out = to_values[idx]
    print('Check method_bruteforce :', np.all(out == array/2)) # Just checking
print('Time method_bruteforce :', timeit.timeit(method_bruteforce, number = 1))

def method_searchsort():
    sort_idx = np.argsort(from_values)
    idx = np.searchsorted(from_values,array,sorter = sort_idx)
    out = to_values[sort_idx][idx]
    print('Check method_searchsort :', np.all(out == array/2)) # Just checking
print('Time method_searchsort :', timeit.timeit(method_searchsort, number = 1))

我得到了以下结果:

Check method_for_loop : True
Time method_for_loop : 2.6411612760275602

Check method_list_comprehension : True
Time method_list_comprehension : 0.07994363596662879

Check method_bruteforce : True
Time method_bruteforce : 11.960559037979692

Check method_searchsort : True
Time method_searchsort : 0.03770717792212963

“searchsort”方法比“for”循环快 一百倍,比 numpy bruteforce 方法快大约 3600 倍。 列表推导方法也是代码简单性和速度之间的一个很好的权衡。

【讨论】:

  • 咳咳……一百次?
  • @PietroBattiston 正确,对此感到抱歉。
【解决方案5】:

没有在数组上没有 python 循环的情况下发布任何解决方案(除了 Celil 的那个,它假设数字是“小”),所以这里有一个替代方案:

def replace(arr, rep_dict):
    """Assumes all elements of "arr" are keys of rep_dict"""

    # Removing the explicit "list" breaks python3
    rep_keys, rep_vals = array(list(zip(*sorted(rep_dict.items()))))

    idces = digitize(arr, rep_keys, right=True)
    # Notice rep_keys[digitize(arr, rep_keys, right=True)] == arr

    return rep_vals[idces]

“idces”的创建方式来自here。

【讨论】:

    【解决方案6】:

    numpy_indexed 包(免责声明:我是它的作者)为此类问题提供了一种优雅而高效的矢量化解决方案:

    import numpy_indexed as npi
    remapped_array = npi.remap(theArray, list(dict.keys()), list(dict.values()))
    

    实现的方法类似于 Jean Lescut 提到的基于搜索排序的方法,但更通用。例如,数组的项目不需要是整数,而是可以是任何类型,甚至是 nd-subarrays 本身;但它应该达到同样的性能。

    【讨论】:

    • 你能不能修改这个单行代码来实现用其他东西替换原始列表中不构成字典键的元素,例如常量? (而不是保留原始值)
    • 不是当前版本的包,但是remap函数有一个'missing' kwarg,如果你打开函数的源代码,你会看到添加这种行为确实会简单。我会在即将发布的版本中这样做;在此之前,如果您愿意,可以随意复制粘贴源代码。
    【解决方案7】:

    使用np.in1d 和np.searchsorted 的完全矢量化解决方案:

    replace = numpy.array([list(replace.keys()), list(replace.values())])    # Create 2D replacement matrix
    mask = numpy.in1d(data, replace[0, :])                                   # Find elements that need replacement
    data[mask] = replace[1, numpy.searchsorted(replace[0, :], data[mask])]   # Replace elements
    

    【讨论】:

      【解决方案8】:
      for i in xrange(len(the_array)):
          the_array[i] = the_dict.get(the_array[i], the_array[i])
      

      【讨论】:

        【解决方案9】:

        好吧,你需要通过theArray,如果它在字典中,则为每个元素替换它。

        for i in xrange( len( theArray ) ):
            if foo[ i ] in dict:
                foo[ i ] = dict[ foo[ i ] ]
        

        【讨论】:

        • 最好将len(theArray)放入变量中,使用xrange。
        • @fuw:是的,xrange,但是将len(theArray) 放入变量中不会有帮助,因为迭代器只计算一次。
        • Py3k 的range 是一个生成器。
        • 目前还没有适用于 Python 3.x 的 NumPy。
        【解决方案10】:

        Pythonic 方式不需要数据是整数,甚至可以是字符串:

        from scipy.stats import rankdata
        import numpy as np
        
        data = np.random.rand(100000)
        replace = {data[0]: 1, data[5]: 8, data[8]: 10}
        
        arr = np.vstack((replace.keys(), replace.values())).transpose()
        arr = arr[arr[:,1].argsort()]
        
        unique = np.unique(data)
        mp = np.vstack((unique, unique)).transpose()
        mp[np.in1d(mp[:,0], arr),1] = arr[:,1]
        data = mp[rankdata(data, 'dense')-1][:,1]
        

        【讨论】:

          猜你喜欢
          • 2012-11-15
          • 2011-12-21
          • 2019-12-25
          • 2019-10-19
          • 2013-09-11
          • 2016-12-29
          • 2019-09-26
          • 2012-05-07
          相关资源
          最近更新 更多