避免使用循环替换NumPy数组中的元素

Question

我有一个很大的1d numpy数组Xold，其中包含给定值。根据2d numpy数组Y指定的规则，这些值将被替换： 例如：

Xold=np.array([0,1,2,3,4])
Y=np.array([[0,0],[1,100],[3,300],[4,400],[2,200]])

每当Xold中的一个值与Y[:,0]中的一个值相同时，Xnew中的新值应该是Y[:,1]中对应的值。这可以通过两个嵌套的for循环实现：

Xnew=np.zeros(len(Xold))
for i in range(len(Xold)):
for j in range(len(Y)):
    if Xold[i]==Y[j,0]:
        Xnew[i]=Y[j,1]

通过给出的示例，这将产生Xnew=[0,100,200,300,400]。然而，对于大型数据集，这个过程非常缓慢。有什么更快更优雅的方法来完成这个任务吗？

- Jann

Answer 1

选择最快的方法

对于替换numpy数组中的元素，这个问题的答案提供了许多不错的方法。现在我们来看看哪一个是最快的。

简而言之： Numpy索引是最快的。

 def meth1(): # suggested by @Slam
    for old, new in Y:  
        Xold[Xold == old] = new

 def meth2(): # suggested by myself, convert y_dict = dict(Y) first
     [y_dict[i] if i in y_dict.keys() else i for i in Xold]

 def meth3(): # suggested by @Eelco Hoogendoom, import numpy_index as npi first
     npi.remap(Xold, keys=Y[:, 0], values=Y[:, 1])

 def meth4(): # suggested by @Brad Solomon, import pandas as pd first 
     pd.Series(Xold).map(pd.Series(Y[:, 1], index=Y[:, 0])).values

  # suggested by @jdehesa. create Xnew = Xold.copy() and index
  # idx = np.searchsorted(Xold, Y[:, 0]) first
  def meth5():             
     Xnew[idx] = Y[:, 1]

结果并不令人意外

 In [39]: timeit.timeit(meth1, number=1000000)                                                                      
 Out[39]: 12.08

 In [40]: timeit.timeit(meth2, number=1000000)                                                                      
 Out[40]: 2.87

 In [38]: timeit.timeit(meth3, number=1000000)                                                                      
 Out[38]: 55.39

 In [12]: timeit.timeit(meth4, number=1000000)                                                                                      
 Out[12]: 256.84

 In [50]: timeit.timeit(meth5, number=1000000)                                                                                      
 Out[50]: 1.12

所以，传统的列表解析是第二快的方法，而获胜的方法是将numpy索引与searchsorted()结合使用。

- Daniel Kislyuk

---

你在测试哪个数据集？ - Divakar

Xold = np.array([0,1,2,3,4,4,4,0]), Y = np.array([[0,0],[1,100],[3,300],[4,400],[2,200]]) - Daniel Kislyuk

Answer 2

我们可以使用 np.searchsorted 来处理一般情况，即当 Y 的第一列数据不一定是有序的时候。

sidx = Y[:,0].argsort()
out = Y[sidx[np.searchsorted(Y[:,0], Xold, sorter=sidx)],1]

样例运行 -

In [53]: Xold
Out[53]: array([14, 10, 12, 13, 11])

In [54]: Y
Out[54]: 
array([[ 10,   0],
       [ 11, 100],
       [ 13, 300],
       [ 14, 400],
       [ 12, 200]])

In [55]: sidx = Y[:,0].argsort()
    ...: out = Y[sidx[np.searchsorted(Y[:,0], Xold, sorter=sidx)],1]

In [56]: out
Out[56]: array([400,   0, 200, 300, 100])

如果并非所有元素都有相应的映射可用，则我们需要做更多的工作，方法如下 -

sidx = Y[:,0].argsort()
sorted_indx = np.searchsorted(Y[:,0], Xold, sorter=sidx)
sorted_indx[sorted_indx==len(sidx)] = len(sidx)-1
idx_out = sidx[sorted_indx]
out = Y[idx_out,1]
out[Y[idx_out,0]!=Xold] = 0 # NA values as 0s

- Divakar

---

这很好，尽管假设存在某些值没有映射的情况，我不确定它们应该设置为0/NA/...还是保留在Xold中。但我认为这只意味着将最后一个0替换为Xold[Y[idx_out,0]!=Xold]，所以无论如何都是一个好的解决方案。 - jdehesa

@jdehesa OP的代码中有Xnew=np.zeros(len(Xold))，所以我觉得使用相同的代码是有道理的。 - Divakar

这段代码对我不起作用： In [16]: sidx = Y[:,0].argsort()
In [17]: out = Y[sidx[np.searchsorted(Y[:,0], Xold, sorter=sidx)],1]
IndexError: index 5 is out of bounds for axis 1 with size 5 - Daniel Kislyuk

正如 @MihaiAlexandruIonut 指出的那样，这是因为我的示例中的 Xold 包含了 Y 中缺失的元素。然而，并没有最初的限制说明这种情况是不允许的。 - Daniel Kislyuk

@DanielKislyuk，因此，这是我帖子末尾的通用解决方案。 - Divakar

Answer 3

以下是一种可能的解决方案：

import numpy as np

Xold = np.array([0, 1, 2, 3, 4])
Y = np.array([[0, 0], [1, 100], [3, 300], [4, 400], [2, 200]])
# Check every X value against every Y first value
m = Xold == Y[:, 0, np.newaxis]
# Check which elements in X are among Y first values
# (so values that are not in Y are not replaced)
m_X = np.any(m, axis=0)
# Compute replacement
# Xold * (1 - m_X) are the non-replaced values
# np.sum(Y[:, 1, np.newaxis] * m, axis=0) * m_X are the replaced values
Xnew = Xold * (1 - m_X) + np.sum(Y[:, 1, np.newaxis] * m, axis=0) * m_X
print(Xnew)

输出：

[  0 100 200 300 400]

这种方法适用于几乎所有情况（未排序的数组，X中多个值的重复，未替换X中的值，Y中没有替换X中任何东西的值），除非您为Y中的同一值提供了两个替换选项，否则将是错误的。然而，它的时间和空间复杂度是X和Y大小的乘积。如果您的问题有其他约束条件（数据已排序，没有重复等），可能可以做得更好。例如，如果X已排序且没有重复元素，并且Y中的每个值都替换X中的一个值（就像您的示例中那样），那么这可能会更快：

import numpy as np

Xold = np.array([0, 1, 2, 3, 4])
Y = np.array([[0, 0], [1, 100], [3, 300], [4, 400], [2, 200]])
idx = np.searchsorted(Xold, Y[:, 0])
Xnew = Xold.copy()
Xnew[idx] = Y[:, 1]
print(Xnew)
# [  0 100 200 300 400]

- jdehesa

Answer 4

首先，您可以使用numpy索引来进行改进，但仍然需要一个循环：

for old, new in Y: 
    Xold[Xold == old] = new

- Slam

---

这是不正确的。对于我的情况 X = np.array([0,1,2,3,4,4]); Y = np.array([[0,1],[1,2],[3,300],[4,400],[2,200]])，它返回 [200 200 200 300 400 400]。 - mathfux

Answer 5

你可以将切片功能与argsort方法结合使用。

Xnew = Y[Y[:,1].argsort()][:, 1][Xold] 

输出

array([  0, 100, 200, 300, 400])

- Mihai Alexandru-Ionut

---

这段代码对我没用。Xnew = Y[Y[:,1].argsort()][:, 1][Xold] IndexError: index 100 is out of bounds for axis 1 with size 5 - Daniel Kislyuk

@DanielKislyuk，这是因为您的“Xold”数组包含了在“Y”数组中不存在的索引。 - Mihai Alexandru-Ionut

是的。你能指出规定了 Y[Y[:,1].argsort()][:, 1][Xold] 替换如何工作的文档吗？我无法理解它。 - Daniel Kislyuk

Answer 6

使用 pd.Series.map() 的解决方案

如果您愿意使用Pandas库，您也可以使用 .map() 以向量化的方式完成此操作：

>>> import pandas as pd
>>> pd.Series(Xold).map(pd.Series(Y[:, 1], index=Y[:, 0]))                                                                                                                                                                    
0      0
1    100
2    200
3    300
4    400
dtype: int64

>>> pd.Series(Xold).map(pd.Series(Y[:, 1], index=Y[:, 0])).values                                                                                                                                                            
array([  0, 100, 200, 300, 400])

对于签名a.map(b)，a在b的索引中寻找其对应的条目，并映射到b中的相应值。

此处的b是pd.Series(Y[:, 1], index=Y[:, 0])，它使用第0列作为索引，并使用第1列作为需要映射到的值。

---

直接使用 pandas.core.algorithms

在幕后, 这将使用 .get_indexer() 和 Cython 实现的 take_1d():

indexer = mapper.index.get_indexer(values)
new_values = algorithms.take_1d(mapper._values, indexer)

如果数组非常大，您可以像这样减少一些开销：

from pandas.core import algorithms

indexer = pd.Index(Y[:, 0]).get_indexer(Xold)  
mapped = algorithms.take_1d(Y[:, 1], indexer)

- Brad Solomon

Answer 7

numpy_indexed 包（免责声明：我是它的作者）包含一个高效的向量化函数，可以解决一般性问题：

import numpy_indexed as npi
Xnew = npi.remap(Xold, keys=Y[:, 0], values=Y[:, 1])

也就是说，这适用于任何dtype，或者当要替换的键和值本身是ndarrays时，并且您可以使用kwarg指定如何处理缺少的元素。

不确定它在性能方面与pandas相比如何；但是这个库中的一个设计选择是，执行像这样的基本操作（或进行分组等操作）不应涉及创建像Series或Table这样的整个新数据类型，这总是让我对使用pandas进行此类操作感到困扰。

- Eelco Hoogendoorn

Answer 8

你可以使用 y = dict(Y) 将 Y 转换为字典，然后运行以下列表推导式。

[y[i] if i in y.keys() else i for i in Xold]

- Daniel Kislyuk