NumPy-np.searchsorted for 2-D arrays

Oct 05 2020

np.searchsorted 1D 배열에만 해당됩니다.

I는이 전적으로 정렬 2 번째의 1 번째 행의 값을 동일한 값도 정렬 0 행째는 1 행째의 대응하는 요소의 동일 값에 대해 다음, 0 번째의 열이 정렬되는 것을 의미하는 2 차원 배열을 행도 정렬됩니다. 즉, 열로 구성된 튜플이 정렬됩니다.

첫 번째 2D 배열에 올바른 열 위치에 삽입해야하는 튜플 열이있는 다른 2D 배열이 있습니다. 1D의 경우 np.searchsorted일반적으로 올바른 위치를 찾기 위해 사용되었습니다.

그러나 2D 배열의 경우 대안이 np.searchsorted있습니까? np.lexsort 가 1D np.argsort 의 2D 대안 인 방법과 유사 합니다.

그러한 기능이 없다면 기존의 numpy 함수를 사용하여 효율적인 방식으로이 기능을 구현할 수 있습니까?

.NET을 dtype포함한 모든 배열에 대한 효율적인 솔루션에 관심이 있습니다 np.object_.

모든 dtype경우 를 처리하는 한 가지 순진한 방법 은 두 배열의 각 열을 1D 배열 (또는 튜플)로 변환 한 다음이 열을 다른 1D 배열로 저장하는 것입니다 dtype = np.object_. 아마도 순진하지 않고 특히 기둥이 상당히 높은 경우 더 빠를 수 있습니다.

답변

1 MadPhysicist Oct 06 2020 at 04:52

여기서 두 가지가 도움이 될 수 있습니다. (1) 구조화 된 배열을 정렬하고 검색 할 수 있습니다. (2) 정수로 매핑 할 수있는 유한 컬렉션이있는 경우이를 유용하게 사용할 수 있습니다.

1D로보기

삽입하려는 문자열 배열이 있다고 가정 해 보겠습니다.

data = np.array([['a', '1'], ['a', 'z'], ['b', 'a']], dtype=object)

배열은 비정형이 아니므로 행 크기 인 dtype을 생성 할 수 있습니다.

dt = np.dtype([('', data.dtype)] * data.shape[1])

여기에 내 뻔뻔스럽게 연결된 대답을 사용 하면 원래 2D 배열을 1D로 볼 수 있습니다.

view = np.ndarray(data.shape[:1], dtype=dt, buffer=data)

이제 검색은 완전히 간단하게 수행 할 수 있습니다.

key = np.array([('a', 'a')], dtype=dt)
index = np.searchsorted(view, key)

적절한 최소값을 사용하여 불완전한 요소의 삽입 인덱스를 찾을 수도 있습니다. 문자열의 경우 이것은 ''.

더 빠른 비교

dtype의 각 필드를 확인할 필요가 없으면 비교에서 더 나은 마일리지를 얻을 수 있습니다. 단일 동종 필드로 유사한 dtype을 만들 수 있습니다.

dt2 = np.dtype([('row', data.dtype, data.shape[1])])

뷰 구성은 이전과 동일합니다.

view = np.ndarray(data.shape[:1], dtype=dt2, buffer=data)

이번에는 키가 약간 다르게 수행됩니다 ( 여기에 다른 플러그 ).

key = np.array([(['a', 'a'],)], dtype=dt2)

객체에 부과 된 정렬 순서는 다음 방법으로 올바르지 않습니다 . 사용자 정의 dtype을 사용하여 행별로 객체 배열을 정렬 합니다. 연결된 질문에 수정 사항이있는 경우 여기에 참조를 남겨 둡니다. 또한 정수 정렬에 여전히 매우 유용합니다.

정수 매핑

검색 할 개체 수가 한정되어있는 경우이를 정수로 매핑하는 것이 더 쉽습니다.

idata = np.empty(data.shape, dtype=int)
keys = [None] * data.shape[1]     # Map index to key per column
indices = [None] * data.shape[1]  # Map key to index per column
for i in range(data.shape[1]):
    keys[i], idata[:, i] = np.unique(data[:, i], return_inverse=True)
    indices[i] = {k: i for i, k in enumerate(keys[i])}  # Assumes hashable objects

idt = np.dtype([('row', idata.dtype, idata.shape[1])])
view = idata.view(idt).ravel()

이것은 data실제로 각 열에 가능한 모든 키를 포함하는 경우에만 작동합니다 . 그렇지 않으면 다른 방법으로 정방향 및 역방향 매핑을 가져와야합니다. 일단 설정되면 키 설정이 훨씬 간단하고 다음 만 필요합니다 indices.

key = np.array([index[k] for index, k in zip(indices, ['a', 'a'])])

추가 개선

가지고있는 카테고리의 수가 8 개 이하이고 각 카테고리에 256 개 이하의 요소가있는 경우 모든 항목을 단일 np.uint64요소 에 맞추면 더 나은 해시를 구성 할 수 있습니다 .

k = math.ceil(math.log(data.shape[1], 2))  # math.log provides base directly
assert 0 < k <= 64
idata = np.empty((data.shape[:1], k), dtype=np.uint8)
...
idata = idata.view(f'>u{k}').ravel()

키도 비슷하게 만들어집니다.

key = np.array([index[k] for index, k in zip(indices, ['a', 'a'])]).view(f'>u{k}')

타이밍

무작위로 섞인 문자열을 사용하여 여기에 표시된 방법 (다른 답변이 아님)의 시간을 정했습니다. 주요 타이밍 매개 변수는 다음과 같습니다.

  • M: 행 수 : 10 ** {2, 3, 4, 5}
  • N: 열 수 : 2 ** {3, 4, 5, 6}
  • K: 삽입 할 요소 수 : 1, 10, M // 10
  • 방법 : individual_fields, combined_field, int_mapping, int_packing. 아래에 표시된 기능.

마지막 두 가지 방법에서는 데이터를 매핑 된 dtype으로 미리 변환하지만 검색 키는 변환하지 않는다고 가정합니다. 따라서 변환 된 데이터를 전달하지만 키 변환 타이밍을 지정합니다.

import numpy as np
from math import ceil, log

def individual_fields(data, keys):
    dt = [('', data.dtype)] * data.shape[1]
    dview = np.ndarray(data.shape[:1], dtype=dt, buffer=data)
    kview = np.ndarray(keys.shape[:1], dtype=dt, buffer=keys)
    return np.searchsorted(dview, kview)

def combined_fields(data, keys):
    dt = [('row', data.dtype, data.shape[1])]
    dview = np.ndarray(data.shape[:1], dtype=dt, buffer=data)
    kview = np.ndarray(keys.shape[:1], dtype=dt, buffer=keys)
    return np.searchsorted(dview, kview)

def int_mapping(idata, keys, indices):
    idt = np.dtype([('row', idata.dtype, idata.shape[1])])
    dview = idata.view(idt).ravel()
    kview = np.empty(keys.shape[0], dtype=idt)
    for i, (index, key) in enumerate(zip(indices, keys.T)):
        kview['row'][:, i] = [index[k] for k in key]
    return np.searchsorted(dview, kview)

def int_packing(idata, keys, indices):
    idt = f'>u{idata.shape[1]}'
    dview = idata.view(idt).ravel()
    kview = np.empty(keys.shape, dtype=np.uint8)
    for i, (index, key) in enumerate(zip(indices, keys.T)):
        kview[:, i] = [index[k] for k in key]
    kview = kview.view(idt).ravel()
    return np.searchsorted(dview, kview)

타이밍 코드 :

from math import ceil, log
from string import ascii_lowercase
from timeit import Timer

def time(m, n, k, fn, *args):
    t = Timer(lambda: fn(*args))
    s = t.autorange()[0]
    print(f'M={m}; N={n}; K={k} {fn.__name__}: {min(t.repeat(5, s)) / s}')

selection = np.array(list(ascii_lowercase), dtype=object)
for lM in range(2, 6):
    M = 10**lM
    for lN in range(3, 6):
        N = 2**lN
        data = np.random.choice(selection, size=(M, N))
        np.ndarray(data.shape[0], dtype=[('', data.dtype)] * data.shape[1], buffer=data).sort()
        idata = np.array([[ord(a) - ord('a') for a in row] for row in data], dtype=np.uint8)
        ikeys = [selection] * data.shape[1]
        indices = [{k: i for i, k in enumerate(selection)}] * data.shape[1]
        for K in (1, 10, M // 10):
            key = np.random.choice(selection, size=(K, N))
            time(M, N, K, individual_fields, data, key)
            time(M, N, K, combined_fields, data, key)
            time(M, N, K, int_mapping, idata, key, indices)
            if N <= 8:
                time(M, N, K, int_packing, idata, key, indices)

결과 :

M = 100 (단위 = 미국)

   |                           K                           |
   +---------------------------+---------------------------+
N  |             1             |            10             |
   +------+------+------+------+------+------+------+------+
   |  IF  |  CF  |  IM  |  IP  |  IF  |  CF  |  IM  |  IP  |
---+------+------+------+------+------+------+------+------+
 8 | 25.9 | 18.6 | 52.6 | 48.2 | 35.8 | 22.7 | 76.3 | 68.2 | 
16 | 40.1 | 19.0 | 87.6 |  --  | 51.1 | 22.8 | 130. |  --  |
32 | 68.3 | 18.7 | 157. |  --  | 79.1 | 22.4 | 236. |  --  |
64 | 125. | 18.7 | 290. |  --  | 135. | 22.4 | 447. |  --  |
---+------+------+------+------+------+------+------+------+

M = 1000 (단위 = 미국)

   |                                         K                                         |
   +---------------------------+---------------------------+---------------------------+
N  |             1             |            10             |            100            |
   +------+------+------+------+------+------+------+------+------+------+------+------+
   |  IF  |  CF  |  IM  |  IP  |  IF  |  CF  |  IM  |  IP  |  IF  |  CF  |  IM  |  IP  |
---+------+------+------+------+------+------+------+------+------+------+------+------+
 8 | 26.9 | 19.1 | 55.0 | 55.0 | 44.8 | 25.1 | 79.2 | 75.0 | 218. | 74.4 | 305. | 250. |
16 | 41.0 | 19.2 | 90.5 |  --  | 59.3 | 24.6 | 134. |  --  | 244. | 79.0 | 524. |  --  | 
32 | 68.5 | 19.0 | 159. |  --  | 87.4 | 24.7 | 241. |  --  | 271. | 80.5 | 984. |  --  |
64 | 128. | 19.7 | 312. |  --  | 168. | 26.0 | 549. |  --  | 396. | 7.78 | 2.0k |  --  |
---+------+------+------+------+------+------+------+------+------+------+------+------+

M = 10K (단위 = 미국)

   |                                         K                                         |
   +---------------------------+---------------------------+---------------------------+
N  |             1             |            10             |           1000            |
   +------+------+------+------+------+------+------+------+------+------+------+------+
   |  IF  |  CF  |  IM  |  IP  |  IF  |  CF  |  IM  |  IP  |  IF  |  CF  |  IM  |  IP  |
---+------+------+------+------+------+------+------+------+------+------+------+------+
 8 | 28.8 | 19.5 | 54.5 | 107. | 57.0 | 27.2 | 90.5 | 128. | 3.2k | 762. | 2.7k | 2.1k |
16 | 42.5 | 19.6 | 90.4 |  --  | 73.0 | 27.2 | 140. |  --  | 3.3k | 752. | 4.6k |  --  |
32 | 73.0 | 19.7 | 164. |  --  | 104. | 26.7 | 246. |  --  | 3.4k | 803. | 8.6k |  --  |
64 | 135. | 19.8 | 302. |  --  | 162. | 26.1 | 466. |  --  | 3.7k | 791. | 17.k |  --  |
---+------+------+------+------+------+------+------+------+------+------+------+------+

individual_fields(IF)는 일반적으로 가장 빠른 작업 방법입니다. 열 수에 비례하여 복잡성이 증가합니다. 불행히도 combined_fields(CF)는 객체 배열에서 작동하지 않습니다. 그렇지 않으면 가장 빠른 방법 일뿐만 아니라 열이 증가해도 복잡성이 증가하지 않는 방법이기도합니다.

파이썬 객체를 키에 매핑하는 것이 느리기 때문에 내가 더 빠를 것이라고 생각한 모든 기술은 그렇지 않습니다 (예를 들어 패킹 된 int 배열의 실제 조회는 구조화 된 배열보다 훨씬 빠릅니다).

참고 문헌

이 코드가 전혀 작동하도록하기 위해 내가 요청해야하는 추가 질문은 다음과 같습니다.

  • 다른 dtype에서 개체 배열보기
  • 배열을 포함하는 단일 구조화 된 요소로 배열 만들기
  • 사용자 지정 dtype을 사용하여 행별로 개체 배열 정렬
1 Arty Oct 05 2020 at 14:33

몇 가지 고급 전략을 만들었습니다.

또한 다른 내 대답tuples 에서 like를 사용하는 간단한 전략 이 구현됩니다.

모든 솔루션의 타이밍이 측정됩니다.

대부분의 전략은 np.searchsorted기본 엔진 으로 사용 됩니다. 이러한 고급 전략을 구현 _CmpIx하기 __lt__위해 np.searchsorted호출 에 대한 사용자 지정 비교 함수 ( ) 를 제공하기 위해 특수 래핑 클래스 가 사용되었습니다 .

  1. py.tuples전략은 모든 열을 튜플로 변환하고 np.object_dtype의 numpy 1D 배열로 저장 한 다음 정기적 인 검색 정렬을 수행합니다.
  2. py.zip 동일한 작업을 느리게 수행하기 위해 Python의 zip을 사용합니다.
  3. np.lexsort전략은 np.lexsort두 열을 사전 식으로 비교하기 위해 사용합니다 .
  4. np.nonzeronp.flatnonzero(a != b)표현을 사용 합니다.
  5. cmp_numba제공된 두 요소의 빠른 사전 적 지연 비교를 위해 래퍼 내부에 미리 컴파일 된 numba 코드를 사용 _CmpIx합니다.
  6. np.searchsorted 표준 numpy의 기능을 사용하지만 1D 경우에만 측정됩니다.
  7. 대한 numba전략 전체 검색 알고리즘을 사용하여 처음부터 구현 Numba의 엔진, 알고리즘을 기반으로 이진 검색 . 이 알고리즘의 변형 이 _py있으며 Numba 컴파일러를 사용하므로 훨씬 빠르며 동일한 알고리즘이지만 컴파일되지 않습니다. 또한 삽입 할 배열의 추가 최적화를 수행하는 플레이버가 이미 정렬되어 있습니다._nm_nm_py_sorted
  8. view1d- 이 답변에서 @MadPhysicist 가 제안한 방법 . 코드에서 주석 처리했습니다. 모든 키 길이> 1에 대한 대부분의 테스트에 대해 잘못된 답변을 반환했기 때문입니다. 아마도 배열에 대한 원시보기 문제 때문일 것입니다.

온라인으로 시도하십시오!

class SearchSorted2D:
    class _CmpIx:
        def __init__(self, t, p, i):
            self.p, self.i = p, i
            self.leg = self.leg_cache()[t]
            self.lt = lambda o: self.leg(self, o, False) if self.i != o.i else False
            self.le = lambda o: self.leg(self, o, True) if self.i != o.i else True
        @classmethod
        def leg_cache(cls):
            if not hasattr(cls, 'leg_cache_data'):
                cls.leg_cache_data = {
                    'py.zip': cls._leg_py_zip, 'np.lexsort': cls._leg_np_lexsort,
                    'np.nonzero': cls._leg_np_nonzero, 'cmp_numba': cls._leg_numba_create(),
                }
            return cls.leg_cache_data
        def __eq__(self, o): return not self.lt(o) and self.le(o)
        def __ne__(self, o): return self.lt(o) or not self.le(o)
        def __lt__(self, o): return self.lt(o)
        def __le__(self, o): return self.le(o)
        def __gt__(self, o): return not self.le(o)
        def __ge__(self, o): return not self.lt(o)
        @staticmethod
        def _leg_np_lexsort(self, o, eq):
            import numpy as np
            ia, ib = (self.i, o.i) if eq else (o.i, self.i)
            return (np.lexsort(self.p.ab[::-1, ia : (ib + (-1, 1)[ib >= ia], None)[ib == 0] : ib - ia])[0] == 0) == eq
        @staticmethod
        def _leg_py_zip(self, o, eq):
            for l, r in zip(self.p.ab[:, self.i], self.p.ab[:, o.i]):
                if l < r:
                    return True
                if l > r:
                    return False
            return eq
        @staticmethod
        def _leg_np_nonzero(self, o, eq):
            import numpy as np
            a, b = self.p.ab[:, self.i], self.p.ab[:, o.i]
            ix = np.flatnonzero(a != b)
            return a[ix[0]] < b[ix[0]] if ix.size != 0 else eq
        @staticmethod
        def _leg_numba_create():
            import numpy as np

            try:
                from numba.pycc import CC
                cc = CC('ss_numba_mod')
                @cc.export('ss_numba_i8', 'b1(i8[:],i8[:],b1)')
                def ss_numba(a, b, eq):
                    for i in range(a.size):
                        if a[i] < b[i]:
                            return True
                        elif b[i] < a[i]:
                            return False
                    return eq
                cc.compile()
                success = True
            except:    
                success = False
                
            if success:
                try:
                    import ss_numba_mod
                except:
                    success = False
            
            def odo(self, o, eq):
                a, b = self.p.ab[:, self.i], self.p.ab[:, o.i]
                assert a.ndim == 1 and a.shape == b.shape, (a.shape, b.shape)
                return ss_numba_mod.ss_numba_i8(a, b, eq)
                
            return odo if success else None

    def __init__(self, type_):
        import numpy as np
        self.type_ = type_
        self.ci = np.array([], dtype = np.object_)
    def __call__(self, a, b, *pargs, **nargs):
        import numpy as np
        self.ab = np.concatenate((a, b), axis = 1)
        self._grow(self.ab.shape[1])
        ix = np.searchsorted(self.ci[:a.shape[1]], self.ci[a.shape[1] : a.shape[1] + b.shape[1]], *pargs, **nargs)
        return ix
    def _grow(self, to):
        import numpy as np
        if self.ci.size >= to:
            return
        import math
        to = 1 << math.ceil(math.log(to) / math.log(2))
        self.ci = np.concatenate((self.ci, [self._CmpIx(self.type_, self, i) for i in range(self.ci.size, to)]))

class SearchSorted2DNumba:
    @classmethod
    def do(cls, a, v, side = 'left', *, vsorted = False, numba_ = True):
        import numpy as np

        if not hasattr(cls, '_ido_numba'):
            def _ido_regular(a, b, vsorted, lrt):
                nk, na, nb = a.shape[0], a.shape[1], b.shape[1]
                res = np.zeros((2, nb), dtype = np.int64)
                max_depth = 0
                if nb == 0:
                    return res, max_depth
                #lb, le, rb, re = 0, 0, 0, 0
                lrb, lre = 0, 0
                
                if vsorted:
                    brngs = np.zeros((nb, 6), dtype = np.int64)
                    brngs[0, :4] = (-1, 0, nb >> 1, nb)
                    i, j, size = 0, 1, 1
                    while i < j:
                        for k in range(i, j):
                            cbrng = brngs[k]
                            bp, bb, bm, be = cbrng[:4]
                            if bb < bm:
                                brngs[size, :4] = (k, bb, (bb + bm) >> 1, bm)
                                size += 1
                            bmp1 = bm + 1
                            if bmp1 < be:
                                brngs[size, :4] = (k, bmp1, (bmp1 + be) >> 1, be)
                                size += 1
                        i, j = j, size
                    assert size == nb
                    brngs[:, 4:] = -1

                for ibc in range(nb):
                    if not vsorted:
                        ib, lrb, lre = ibc, 0, na
                    else:
                        ibpi, ib = int(brngs[ibc, 0]), int(brngs[ibc, 2])
                        if ibpi == -1:
                            lrb, lre = 0, na
                        else:
                            ibp = int(brngs[ibpi, 2])
                            if ib < ibp:
                                lrb, lre = int(brngs[ibpi, 4]), int(res[1, ibp])
                            else:
                                lrb, lre = int(res[0, ibp]), int(brngs[ibpi, 5])
                        brngs[ibc, 4 : 6] = (lrb, lre)
                        assert lrb != -1 and lre != -1
                        
                    for ik in range(nk):
                        if lrb >= lre:
                            if ik > max_depth:
                                max_depth = ik
                            break

                        bv = b[ik, ib]
                        
                        # Binary searches
                        
                        if nk != 1 or lrt == 2:
                            cb, ce = lrb, lre
                            while cb < ce:
                                cm = (cb + ce) >> 1
                                av = a[ik, cm]
                                if av < bv:
                                    cb = cm + 1
                                elif bv < av:
                                    ce = cm
                                else:
                                    break
                            lrb, lre = cb, ce
                                
                        if nk != 1 or lrt >= 1:
                            cb, ce = lrb, lre
                            while cb < ce:
                                cm = (cb + ce) >> 1
                                if not (bv < a[ik, cm]):
                                    cb = cm + 1
                                else:
                                    ce = cm
                            #rb, re = cb, ce
                            lre = ce
                                
                        if nk != 1 or lrt == 0 or lrt == 2:
                            cb, ce = lrb, lre
                            while cb < ce:
                                cm = (cb + ce) >> 1
                                if a[ik, cm] < bv:
                                    cb = cm + 1
                                else:
                                    ce = cm
                            #lb, le = cb, ce
                            lrb = cb
                            
                        #lrb, lre = lb, re
                            
                    res[:, ib] = (lrb, lre)
                    
                return res, max_depth

            cls._ido_regular = _ido_regular
            
            import numba
            cls._ido_numba = numba.jit(nopython = True, nogil = True, cache = True)(cls._ido_regular)
            
        assert side in ['left', 'right', 'left_right'], side
        a, v = np.array(a), np.array(v)
        assert a.ndim == 2 and v.ndim == 2 and a.shape[0] == v.shape[0], (a.shape, v.shape)
        res, max_depth = (cls._ido_numba if numba_ else cls._ido_regular)(
            a, v, vsorted, {'left': 0, 'right': 1, 'left_right': 2}[side],
        )
        return res[0] if side == 'left' else res[1] if side == 'right' else res

def Test():
    import time
    import numpy as np
    np.random.seed(0)
    
    def round_float_fixed_str(x, n = 0):
        if type(x) is int:
            return str(x)
        s = str(round(float(x), n))
        if n > 0:
            s += '0' * (n - (len(s) - 1 - s.rfind('.')))
        return s

    def to_tuples(x):
        r = np.empty([x.shape[1]], dtype = np.object_)
        r[:] = [tuple(e) for e in x.T]
        return r
    
    searchsorted2d = {
        'py.zip': SearchSorted2D('py.zip'),
        'np.nonzero': SearchSorted2D('np.nonzero'),
        'np.lexsort': SearchSorted2D('np.lexsort'),
        'cmp_numba': SearchSorted2D('cmp_numba'),
    }
    
    for iklen, klen in enumerate([1, 1, 2, 5, 10, 20, 50, 100, 200]):
        times = {}
        for side in ['left', 'right']:
            a = np.zeros((klen, 0), dtype = np.int64)
            tac = to_tuples(a)

            for itest in range((15, 100)[iklen == 0]):
                b = np.random.randint(0, (3, 100000)[iklen == 0], (klen, np.random.randint(1, (1000, 2000)[iklen == 0])), dtype = np.int64)
                b = b[:, np.lexsort(b[::-1])]
                
                if iklen == 0:
                    assert klen == 1, klen
                    ts = time.time()
                    ix1 = np.searchsorted(a[0], b[0], side = side)
                    te = time.time()
                    times['np.searchsorted'] = times.get('np.searchsorted', 0.) + te - ts
                    
                for cached in [False, True]:
                    ts = time.time()
                    tb = to_tuples(b)
                    ta = tac if cached else to_tuples(a)
                    ix1 = np.searchsorted(ta, tb, side = side)
                    if not cached:
                        ix0 = ix1
                    tac = np.insert(tac, ix0, tb) if cached else tac
                    te = time.time()
                    timesk = f'py.tuples{("", "_cached")[cached]}'
                    times[timesk] = times.get(timesk, 0.) + te - ts

                for type_ in searchsorted2d.keys():
                    if iklen == 0 and type_ in ['np.nonzero', 'np.lexsort']:
                        continue
                    ss = searchsorted2d[type_]
                    try:
                        ts = time.time()
                        ix1 = ss(a, b, side = side)
                        te = time.time()
                        times[type_] = times.get(type_, 0.) + te - ts
                        assert np.array_equal(ix0, ix1)
                    except Exception:
                        times[type_ + '!failed'] = 0.

                for numba_ in [False, True]:
                    for vsorted in [False, True]:
                        if numba_:
                            # Heat-up/pre-compile numba
                            SearchSorted2DNumba.do(a, b, side = side, vsorted = vsorted, numba_ = numba_)
                        
                        ts = time.time()
                        ix1 = SearchSorted2DNumba.do(a, b, side = side, vsorted = vsorted, numba_ = numba_)
                        te = time.time()
                        timesk = f'numba{("_py", "_nm")[numba_]}{("", "_sorted")[vsorted]}'
                        times[timesk] = times.get(timesk, 0.) + te - ts
                        assert np.array_equal(ix0, ix1)


                # View-1D methods suggested by @MadPhysicist
                if False: # Commented out as working just some-times
                    aT, bT = np.copy(a.T), np.copy(b.T)
                    assert aT.ndim == 2 and bT.ndim == 2 and aT.shape[1] == klen and bT.shape[1] == klen, (aT.shape, bT.shape, klen)
                    
                    for ty in ['if', 'cf']:
                        try:
                            dt = np.dtype({'if': [('', b.dtype)] * klen, 'cf': [('row', b.dtype, klen)]}[ty])
                            ts = time.time()
                            va = np.ndarray(aT.shape[:1], dtype = dt, buffer = aT)
                            vb = np.ndarray(bT.shape[:1], dtype = dt, buffer = bT)
                            ix1 = np.searchsorted(va, vb, side = side)
                            te = time.time()
                            assert np.array_equal(ix0, ix1), (ix0.shape, ix1.shape, ix0[:20], ix1[:20])
                            times[f'view1d_{ty}'] = times.get(f'view1d_{ty}', 0.) + te - ts
                        except Exception:
                            raise
                
                a = np.insert(a, ix0, b, axis = 1)
            
        stimes = ([f'key_len: {str(klen).rjust(3)}'] +
            [f'{k}: {round_float_fixed_str(v, 4).rjust(7)}' for k, v in times.items()])
        nlines = 4
        print('-' * 50 + '\n' + ('', '!LARGE!:\n')[iklen == 0], end = '')
        for i in range(nlines):
            print(',  '.join(stimes[len(stimes) * i // nlines : len(stimes) * (i + 1) // nlines]), flush = True)
            
Test()

출력 :

--------------------------------------------------
!LARGE!:
key_len:   1,  np.searchsorted:  0.0250
py.tuples_cached:  3.3113,  py.tuples: 30.5263,  py.zip: 40.9785
cmp_numba: 25.7826,  numba_py:  3.6673
numba_py_sorted:  6.8926,  numba_nm:  0.0466,  numba_nm_sorted:  0.0505
--------------------------------------------------
key_len:   1,  py.tuples_cached:  0.1371
py.tuples:  0.4698,  py.zip:  1.2005,  np.nonzero:  4.7827
np.lexsort:  4.4672,  cmp_numba:  1.0644,  numba_py:  0.2748
numba_py_sorted:  0.5699,  numba_nm:  0.0005,  numba_nm_sorted:  0.0020
--------------------------------------------------
key_len:   2,  py.tuples_cached:  0.1131
py.tuples:  0.3643,  py.zip:  1.0670,  np.nonzero:  4.5199
np.lexsort:  3.4595,  cmp_numba:  0.8582,  numba_py:  0.4958
numba_py_sorted:  0.6454,  numba_nm:  0.0025,  numba_nm_sorted:  0.0025
--------------------------------------------------
key_len:   5,  py.tuples_cached:  0.1876
py.tuples:  0.4493,  py.zip:  1.6342,  np.nonzero:  5.5168
np.lexsort:  4.6086,  cmp_numba:  1.0939,  numba_py:  1.0607
numba_py_sorted:  0.9737,  numba_nm:  0.0050,  numba_nm_sorted:  0.0065
--------------------------------------------------
key_len:  10,  py.tuples_cached:  0.6017
py.tuples:  1.2275,  py.zip:  3.5276,  np.nonzero: 13.5460
np.lexsort: 12.4183,  cmp_numba:  2.5404,  numba_py:  2.8334
numba_py_sorted:  2.3991,  numba_nm:  0.0165,  numba_nm_sorted:  0.0155
--------------------------------------------------
key_len:  20,  py.tuples_cached:  0.8316
py.tuples:  1.3759,  py.zip:  3.4238,  np.nonzero: 13.7834
np.lexsort: 16.2164,  cmp_numba:  2.4483,  numba_py:  2.6405
numba_py_sorted:  2.2226,  numba_nm:  0.0170,  numba_nm_sorted:  0.0160
--------------------------------------------------
key_len:  50,  py.tuples_cached:  1.0443
py.tuples:  1.4085,  py.zip:  2.2475,  np.nonzero:  9.1673
np.lexsort: 19.5266,  cmp_numba:  1.6181,  numba_py:  1.7731
numba_py_sorted:  1.4637,  numba_nm:  0.0415,  numba_nm_sorted:  0.0405
--------------------------------------------------
key_len: 100,  py.tuples_cached:  2.0136
py.tuples:  2.5380,  py.zip:  2.2279,  np.nonzero:  9.2929
np.lexsort: 33.9505,  cmp_numba:  1.5722,  numba_py:  1.7158
numba_py_sorted:  1.4208,  numba_nm:  0.0871,  numba_nm_sorted:  0.0851
--------------------------------------------------
key_len: 200,  py.tuples_cached:  3.5945
py.tuples:  4.1847,  py.zip:  2.3553,  np.nonzero: 11.3781
np.lexsort: 66.0104,  cmp_numba:  1.8153,  numba_py:  1.9449
numba_py_sorted:  1.6463,  numba_nm:  0.1661,  numba_nm_sorted:  0.1651

타이밍 numba_nm구현이 가장 빠르기 때문에 다음으로 빠른 ( py.zip또는 py.tuples_cached) 성능을 시간별로 능가 15-100x합니다. 그리고 1D 케이스의 1.85x표준 과 비슷한 속도 ( 느림)를 제공 np.searchsorted합니다. 또한 _sorted풍미가 상황을 개선하지 않는 것으로 나타났습니다 (예 : 정렬되는 삽입 된 배열에 대한 정보 사용).

cmp_numba기계 코드로 컴파일 된 메서드 는 동일한 알고리즘을 수행하는 1.5x것보다 평균적 으로 약 배 빠르지 py.zip만 순수한 파이썬에서 나타납니다 . 평균 최대 동일 키 깊이가 15-18요소 주변에 있기 때문에 numba는 여기에서 많은 속도를 얻지 못합니다. 깊이가 수백이면 numba 코드는 아마도 엄청난 속도 향상을 가져올 것입니다.

py.tuples_cached전략은 py.zip키 길이의 경우 보다 빠릅니다 <= 100.

또한 np.lexsort실제로 매우 느린 것으로 보입니다. 단 두 개의 열의 경우에 최적화되지 않았거나 행을 목록으로 분할하는 것과 같은 전처리를 수행하는 데 시간을 소비하거나 지연되지 않은 사전 비교를 수행합니다. 마지막 경우는 다음과 같습니다. 아마도 lexsort가 키 길이가 증가함에 따라 느려지는 진짜 이유 일 것입니다.

전략 np.nonzero은 또한 지연되지 않으므로 느리게 작동하고 키 길이 증가에 따라 속도가 느려집니다 (그러나 그렇게 빠르지는 np.lexsort않습니다).

위의 타이밍은 정확하지 않을 수 있습니다. CPU가 과열 될 때마다 임의의 시간에 코어 주파수를 2-2.3 배 느리게하고 노트북 내부의 강력한 CPU이기 때문에 자주 과열되기 때문입니다.

Arty Oct 05 2020 at 01:29

내 질문에서 언급 한 첫 번째 순진한 솔루션을 게시하면 2D 배열을 dtype = np.object_Python 튜플으로 원래 열 을 포함하는 1D 배열로 변환 한 다음 1D를 사용 np.searchsorted하고 솔루션은 모든 dtype. 실제로이 솔루션은 현재 질문에 대한 다른 답변에서 측정했을 때 순진하지 않고 매우 빠릅니다. 특히 키 길이가 100 미만인 경우 빠릅니다.

온라인으로 시도하십시오!

import numpy as np
np.random.seed(0)

def to_obj(x):
    res = np.empty((x.shape[0],), dtype = np.object_)
    res[:] = [tuple(np.squeeze(e, 0)) for e in np.split(x, x.shape[0], axis = 0)]
    return res

a = np.random.randint(0, 3, (10, 23))
b = np.random.randint(0, 3, (10, 15))

a, b = [x[:, np.lexsort(x[::-1])] for x in (a, b)]

print(np.concatenate((np.arange(a.shape[1])[None, :], a)), '\n\n', b, '\n')

a, b = [to_obj(x.T) for x in (a, b)]

print(np.searchsorted(a, b))