代码之家  ›  专栏  ›  技术社区  ›  Ted Petrou

使用Cython查找2D NumPy阵列唯一行的最快方法

  •  3
  • Ted Petrou  · 技术社区  · 8 年前

    我有一个可以是任何类型的2D NumPy数组,但在本例中,我们可以假设它是整数。我正在寻找查找数组中所有唯一行的最快方法。

    我最初的策略是将每一行转换为一个元组,并将其添加到一个集合中。如果集合的长度增加,这意味着找到了唯一的行。

    我不知道如何快速地将每一行散列为字节 . 有一个问题是 entire array is hashed here .

    我所尝试的-元组创建

    有许多方法可以创建元组,每种方法都会影响性能。这是我的函数,我展示了4种不同的变体:

    版本1:

    def unique_int_tuple1(ndarray[np.int64_t, ndim=2] a):
        cdef int i, len_before
        cdef int nr = a.shape[0]
        cdef int nc = a.shape[1]
        cdef set s = set()
        cdef ndarray[np.uint8_t, cast = True] idx = np.zeros(nr, dtype='bool')
    
        for i in range(nr):
            len_before = len(s)
            s.add(tuple(a[i]))        # THIS LINE IS CHANGED FOR ALL VERSIONS
            if len(s) > len_before:
                idx[i] = True
        return idx
    

    版本2:

    s.add(tuple([a[i, j] for j in range(nc)]))
    

    版本3:

    vals 长度等于列数的列表

    for j in range(nc):
        vals[j] = a[i, j]
        s.add(tuple(vals))
    

    版本4:

    s.add((a[i, 0], a[i, 1], a[i, 2], a[i, 3]))
    

    表演

    a = np.random.randint(0, 8, (10**5, 4))
    %timeit unique_int_tuple1(a)
    125 ms ± 1.96 ms per loop (mean ± std. dev. of 7 runs, 10 loops each)
    
    %timeit unique_int_tuple2(a)
    14.5 ms ± 93.9 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)
    
    %timeit unique_int_tuple3(a)
    11.7 ms ± 126 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)
    
    %timeit unique_int_tuple4(a)
    9.59 ms ± 108 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)
    

    避免 tuple 构造函数(版本4)带来了良好的性能增益。

    使用 tostring

    从上面链接的SO问题中,我可以使用 转换为字符串 方法,然后对此进行哈希运算。

    def unique_int_tostring(ndarray[np.int64_t, ndim=2] a):
        cdef int i, j
        cdef int nr = a.shape[0]
        cdef int nc = a.shape[1]
        cdef set s = set()
        cdef ndarray[np.uint8_t, cast = True] idx = np.zeros(nr, dtype='bool')
    
        for i in range(nr):
            len_before = len(s)
            s.add(a[i].tostring())
            if len(s) > len_before:
                idx[i] = True
        return idx
    

    这可以工作,但速度很慢:

    %timeit unique_int_tostring(a)
    40 ms ± 428 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)
    

    使用类型化memoryview

    我认为,经济放缓的很大一部分是每一行的准入 a[i] . 我们可以使用类型化MemoryView来提高性能,但是 我不知道如何将类型化MemoryView的元素转换为字符串 所以可以对它们进行哈希运算。

    def unique_int_memoryview(long[:, :] a):
        cdef int i, j
        cdef int nr = a.shape[0]
        cdef int nc = a.shape[1]
        cdef set s = set()
        for i in range(nr):
            s.add(<SOMETHING>)   # NO IDEA HERE
        return s
    
    2 回复  |  直到 8 年前
        1
  •  3
  •   HYRY    8 年前

    您可以使用 ndarray.view() 更改 dtype 到 byte string ,然后使用 pandas.Series.duplicated() 要查找重复的行,请执行以下操作:

    import numpy as np
    
    a = np.random.randint(0, 5, size=(200, 3))
    s = pd.Series(a.view(("S", a[0].nbytes))[:, 0])
    s.duplicated()
    

    的核心算法 duplicated() 在Cython中实现。但是,它需要将原始数组转换为对象数组,这可能会很慢。

    跳过 object array ,您可以使用 khash library 熊猫直接使用的,下面是C代码:

    #include "khash.h"
    
    typedef struct _Buf{
        unsigned short n;
        char * pdata;
    } Buf;
    
    khint32_t kh_buf_hash_func(Buf key)
    {
        int i;
        char * s;
        khint32_t hash = 0;
        s = key.pdata;
        for(i=0;i<key.n;i++)
        {
            hash += *s++;
            hash += (hash << 10);
            hash ^= (hash >> 6);
        }
        hash += (hash << 3);
        hash ^= (hash >> 11);
        hash += (hash << 15);    
        return hash;
    }
    
    khint32_t kh_buf_hash_equal(Buf a, Buf b)
    {
        int i;
        if(a.n != b.n) return 0;
        for(i=0;i<a.n;i++){
            if(a.pdata[i] != b.pdata[i]) return 0;
        }
        return 1;
    }
    
    KHASH_INIT(buf, Buf, char, 0, kh_buf_hash_func, kh_buf_hash_equal)
    
    
    void duplicated(char * arr, int row_size, int count, char * res)
    {
        kh_buf_t * khbuf;
        Buf row;
        int i, absent;
        khint_t k;
        row.n = row_size;
    
        khbuf = kh_init_buf();
        kh_resize_buf(khbuf, 4 * count);
    
        for(i=0;i<count;i++){
            row.pdata = &arr[i * row_size];
            k = kh_put_buf(khbuf, row, &absent);
            if (absent){
                res[i] = 0;
            }
            else{
                res[i] = 1;
            }
        }    
        kh_destroy_buf(khbuf);
    }
    

    然后将 重复() Cython或Ctypes或cffi函数。

        2
  •  2
  •   chrisb    8 年前

    令我惊讶的是,这个速度较慢,但不管它值多少钱,这里有一个c++解决方案,它可以实现您所指的功能—将每一行作为一组字节进行散列。“诀窍”是获取元素的地址 <char*>&a[i, 0] -其他一切都是簿记。

    我可能做了一些明显的次优和/或使用不同的哈希表impl性能可能更好。

    编辑:

    回复:如何从行创建字符串 我认为你能做的最好的就是——构建一个 bytes 对象。这必然涉及到行的副本参见c api docs .

    %%cython
    from numpy cimport *
    cimport numpy as np
    import numpy as np
    from cpython.bytes cimport PyBytes_FromStringAndSize
    
    def unique_int_string(ndarray[np.int64_t, ndim=2] a):
        cdef int i, len_before
        cdef int nr = a.shape[0]
        cdef int nc = a.shape[1]
        cdef set s = set()
        cdef ndarray[np.uint8_t, cast = True] idx = np.zeros(nr, dtype='bool')
        cdef bytes string
    
        for i in range(nr):
            len_before = len(s)
            string = PyBytes_FromStringAndSize(<char*>&a[i, 0], sizeof(np.int64_t) * nc)
            s.add(string)
            if len(s) > len_before:
                idx[i] = True
        return idx
    

    //时间安排

    In [9]: from unique import unique_ints
    
    In [10]: %timeit unique_int_tuple4(a)
    100 loops, best of 3: 10.1 ms per loop
    
    In [11]: %timeit unique_ints(a)
    100 loops, best of 3: 11.9 ms per loop
    
    In [12]: (unique_ints(a) == unique_int_tuple4(a)).all()
    Out[12]: True
    

    //助手。h类

    #include <unordered_set>
    #include <cstring>
    
    struct Hasher {
        size_t size;
        size_t operator()(char* buf) const {
            // https://github.com/yt-project/yt/blob/c1569367c6e3d8d0a02e10d0f3d0bd701d2e2114/yt/utilities/lib/fnv_hash.pyx
            size_t hash_val = 2166136261;
            for (int i = 0; i < size; ++i) {
                    hash_val ^= buf[i];
                    hash_val *= 16777619;
            }
            return hash_val;
        }
    };
    struct Comparer {
        size_t size;
        bool operator()(char* lhs, char* rhs) const {
            return (std::memcmp(lhs, rhs, size) == 0) ? true : false;
        }
    };
    
    struct ArraySet {
        std::unordered_set<char*, Hasher, Comparer> set;
    
        ArraySet (size_t size) : set(0, Hasher{size}, Comparer{size}) {}
        ArraySet () {}
    
        bool add(char* buf) {
            auto p = set.insert(buf);
            return p.second;
        }
    };
    

    //独一无二。pyx公司

    from numpy cimport int64_t, uint8_t
    import numpy as np
    
    cdef extern from 'helper.h' nogil:
        cdef cppclass ArraySet:
            ArraySet()
            ArraySet(size_t)
            bint add(char*)
    
    
    def unique_ints(int64_t[:, :] a):
        cdef:
            Py_ssize_t i, nr = a.shape[0], nc = a.shape[1]
            ArraySet s = ArraySet(sizeof(int64_t) * nc)
            uint8_t[:] idx = np.zeros(nr, dtype='uint8')
    
            bint found;
    
        for i in range(nr):
            found = s.add(<char*>&a[i, 0])
            if found:
                idx[i] = True
    
        return idx
    

    //设置。py公司

    from setuptools import setup, Extension
    from Cython.Build import cythonize
    import numpy as np
    
    exts = [
      Extension('unique', ['unique.pyx'], language='c++', include_dirs=[np.get_include()])
    ]
    
    setup(name='test', ext_modules=cythonize(exts))