【问题标题】:Fastest way to convert a list of indices to 2D numpy array of ones将索引列表转换为二维 numpy 数组的最快方法
【发布时间】:2019-11-01 18:19:42
【问题描述】:

我有一个索引列表

a = [
  [1,2,4],
  [0,2,3],
  [1,3,4],
  [0,2]]

将其转换为 numpy 数组的最快方法是什么,其中每个索引显示 1 出现的位置?

即我想要的是:

output = array([
  [0,1,1,0,1],
  [1,0,1,1,0],
  [0,1,0,1,1],
  [1,0,1,0,0]])

我事先知道数组的最大大小。我知道我可以遍历每个列表并在每个索引位置插入一个 1,但是有没有更快/矢量化的方法来做到这一点?

我的用例可能有数千行/列,我需要这样做数千次,所以越快越好。

【问题讨论】:

  • 它不容易矢量化,因为a 是一个参差不齐的列表。
  • 我认为这可能不是一个好方法,但只是看看 stackoverflow 智囊团能想出什么:)

标签: python arrays performance numpy


【解决方案1】:

根据您的用例,您可能会考虑使用稀疏矩阵。输入矩阵看起来很像Compressed Sparse Row (CSR) 矩阵。也许像

import numpy as np
from scipy.sparse import csr_matrix
from itertools import accumulate


def ragged2csr(inds):
    offset = len(inds[0])
    lens = [len(x) for x in inds]
    indptr = list(accumulate(lens))
    indptr = np.array([x - offset for x in indptr])
    indices = np.array([val for sublist in inds for val in sublist])
    n = indices.size
    data = np.ones(n)
    return csr_matrix((data, indices, indptr))

同样,如果它适合您的用例,稀疏矩阵将允许元素/屏蔽操作随非零的数量而不是元素的数量(行*列)进行缩放,这可能会带来显着的加速(对于足够稀疏的矩阵)。

另一个很好的 CSR 矩阵介绍是Iterative Methods 的第 3.4 节。在这种情况下,dataaaindicesjaindptria。这种格式还具有在不同的包/库中非常受欢迎的好处。

【讨论】:

    【解决方案2】:

    如何使用数组索引?如果您对输入有更多了解,则可以摆脱必须先转换为线性数组的惩罚。

    import numpy as np
    
    
    def main():
        row_count = 4
        col_count = 5
        a = [[1,2,4],[0,2,3],[1,3,4],[0,2]]
    
        # iterate through each row, concatenate all indices and convert them to linear
    
        # numpy append performs copy even if you don't want it, list append is faster
        b = []
        for row_idx, row in enumerate(a):
            b.append(np.array(row, dtype=np.int64) + (row_idx * col_count))
    
        linear_idxs = np.hstack(b)
        #could skip previous steps if given index inputs well before hand, or in linear index order. 
        c = np.zeros(row_count * col_count)
        c[linear_idxs] = 1
        c = c.reshape(row_count, col_count)
        print(c)
    
    
    if __name__ == "__main__":
        main()
    
    #output
    # [[0. 1. 1. 0. 1.]
    #  [1. 0. 1. 1. 0.]
    #  [0. 1. 0. 1. 1.]
    #  [1. 0. 1. 0. 0.]]
    

    【讨论】:

      【解决方案3】:

      如果您可以并且想要使用Cython,您可以创建一个可读(至少如果您不介意打字)和快速的解决方案。

      这里我使用 Cython 的 IPython 绑定在 Jupyter 笔记本中编译它:

      %load_ext cython
      
      %%cython
      
      cimport cython
      cimport numpy as cnp
      import numpy as np
      
      @cython.boundscheck(False)  # remove this if you cannot guarantee that nrow/ncol are correct
      @cython.wraparound(False)
      cpdef cnp.int_t[:, :] mseifert(list a, int nrow, int ncol):
          cdef cnp.int_t[:, :] out = np.zeros([nrow, ncol], dtype=int)
          cdef list subl
          cdef int row_idx
          cdef int col_idx
          for row_idx, subl in enumerate(a):
              for col_idx in subl:
                  out[row_idx, col_idx] = 1
          return out
      

      为了比较此处介绍的解决方案的性能,我使用了我的库 simple_benchmark

      请注意,这使用对数轴同时显示小型和大型阵列的差异。根据我的基准测试,我的函数实际上是最快的解决方案,但也值得指出的是,所有解决方案都相差不远。

      这是我用于基准测试的完整代码:

      import numpy as np
      from simple_benchmark import BenchmarkBuilder, MultiArgument
      import itertools
      
      b = BenchmarkBuilder()
      
      @b.add_function()
      def pp(a, nrow, ncol):
          sz = np.fromiter(map(len, a), int, nrow)
          out = np.zeros((nrow, ncol), int)
          out[np.arange(nrow).repeat(sz), np.fromiter(itertools.chain.from_iterable(a), int, sz.sum())] = 1
          return out
      
      @b.add_function()
      def ts(a, nrow, ncol):
          out = np.zeros((nrow, ncol), int)
          for i, ix in enumerate(a):
              out[i][ix] = 1
          return out
      
      @b.add_function()
      def u9(a, nrow, ncol):
          out = np.zeros((nrow, ncol), int)
          for i, (x, y) in enumerate(zip(a, out)):
              y[x] = 1
              out[i] = y
          return out
      
      b.add_functions([mseifert])
      
      @b.add_arguments("number of rows/columns")
      def argument_provider():
          for n in range(2, 13):
              ncols = 2**n
              a = [
                  sorted(set(np.random.randint(0, ncols, size=np.random.randint(0, ncols)))) 
                  for _ in range(ncols)
              ]
              yield ncols, MultiArgument([a, ncols, ncols])
      
      r = b.run()
      r.plot()
      

      【讨论】:

      • 鉴于输入的尴尬(从 numpy 的角度来看)格式,我真的很惊讶 Cython 在这里获得的收益如此之少。
      • @PaulPanzer 我也有点惊讶——我认为唯一相关(关于性能)的部分是遍历列表。在您的情况下是itertools.chain.from_iterable,在我的情况下是显式迭代。其他一切,如果基本上只是恒定的开销。
      【解决方案4】:

      这个怎么样:

      ncol = 5
      nrow = len(a)
      out = np.zeros((nrow, ncol), int)
      out[np.arange(nrow).repeat([*map(len,a)]), np.concatenate(a)] = 1
      out
      # array([[0, 1, 1, 0, 1],
      #        [1, 0, 1, 1, 0],
      #        [0, 1, 0, 1, 1],
      #        [1, 0, 1, 0, 0]])
      

      这里是一个 1000x1000 二进制数组的时序,注意我使用了上面的优化版本,见下面的函数pp

      pp 21.717635259992676 ms
      ts 37.10938713003998 ms
      u9 37.32933565042913 ms
      

      产生计时的代码:

      import itertools as it
      import numpy as np
      
      def make_data(n,m):
          I,J = np.where(np.random.random((n,m))<np.random.random((n,1)))
          return [*map(np.ndarray.tolist, np.split(J, I.searchsorted(np.arange(1,n))))]
      
      def pp():
          sz = np.fromiter(map(len,a),int,nrow)
          out = np.zeros((nrow,ncol),int)
          out[np.arange(nrow).repeat(sz),np.fromiter(it.chain.from_iterable(a),int,sz.sum())] = 1
          return out
      
      def ts():
          out = np.zeros((nrow,ncol),int)
          for i, ix in enumerate(a):
              out[i][ix] = 1
          return out
      
      def u9():
          out = np.zeros((nrow,ncol),int)
          for i, (x, y) in enumerate(zip(a, out)):
              y[x] = 1
              out[i] = y
          return out
      
      nrow,ncol = 1000,1000
      a = make_data(nrow,ncol)
      
      from timeit import timeit
      assert (pp()==ts()).all()
      assert (pp()==u9()).all()
      
      print("pp", timeit(pp,number=100)*10, "ms")
      print("ts", timeit(ts,number=100)*10, "ms")
      print("u9", timeit(u9,number=100)*10, "ms")
      

      【讨论】:

      • 看起来,使用几个numpy functins 和map 会更慢(当然如果不尝试就无法确认)
      • @TeshanShanukaJ 您是否暗示您的解决方案更快?你有时间支持它吗?性能取决于数据,IMO 这将很好地扩展(这也是我赞成它的原因)。
      • 我没有。我只是发出警告,因为 OP 要求最快的解决方案。我已经提到我的也不会是最快的。由 OP 来测试时机
      • @TeshanShanukaJ 实际上,在中等大小(比如 1000x1000)的示例中,您的似乎要快一些(~10%)。
      • @TeshanShanukaJ 稍作调整后,我现在的速度提高了约 40%。
      【解决方案5】:

      这可能不是最快的方法。您需要使用大型数组比较这些答案的执行时间,以找出最快的方法。这是我的解决方案

      output = np.zeros((4,5))
      for i, ix in enumerate(a):
          output[i][ix] = 1
      
      # output -> 
      #   array([[0, 1, 1, 0, 1],
      #   [1, 0, 1, 1, 0],
      #   [0, 1, 0, 1, 1],
      #   [1, 0, 1, 0, 0]])
      

      【讨论】:

      • 如果提供实际的时间信息,答案会好 2 倍
      【解决方案6】:

      可能不是最好的方法,但我能想到的唯一方法:

      output = np.zeros((4,5))
      for i, (x, y) in enumerate(zip(a, output)):
          y[x] = 1
          output[i] = y
      print(output)
      

      哪些输出:

      [[ 0.  1.  1.  0.  1.]
       [ 1.  0.  1.  1.  0.]
       [ 0.  1.  0.  1.  1.]
       [ 1.  0.  1.  0.  0.]]
      

      【讨论】:

      • 这非常简洁(比我的尝试漂亮得多),尽管就运行时而言,它看起来与手动编写循环相同?
      • @Spcoggthesecond 然后使用 Paul 的解决方案
      猜你喜欢
      • 1970-01-01
      • 2017-10-29
      • 2021-08-28
      • 2011-07-05
      • 2019-05-22
      • 2011-12-04
      • 1970-01-01
      • 2020-01-29
      • 2011-06-14
      相关资源
      最近更新 更多