[번역] Similarity Search and Similar Image Search in Elasticsearch

유사 이미지 검색 분야 관련 블로그 글을 번역

& 실제 테스트하는 내용의 포스팅입니다.

 

 

 

"원본 링크 [출처] 

https://medium.com/@kumon/how-to-realize-similarity-search-with-elasticsearch-3dd5641b9adb"

 

 

 

 

효율적으로 큰 차원의 벡터 검색을 지원하는 검색 엔진을 관리하면 많은 이점이 존재한다. 이번 포스팅에서 아마존 엘라스틱 서비스에서 knn 기능을 사용하고 평가한다.

 

Introduction

최근, AWS는 다음과 같은 블로그 포스팅을 게시하였다.

Amazon Elasticsearch Service를 사용하여 k-NN(k-최근접 이웃) 유사성 검색 엔진 구축

Amazon Elasticsearch Service를 사용하여 k-NN(k-최근접 이웃) 유사성 검색 엔진 구축

 

이 포스팅에서는, HNSW의 하나의 실행방식인 NMS 라이브러리와 함께 경량의 유사 검색을 지원한다고 하였다. kNN을 가능하게 하는 이 플러그인은 엘라스틱 서치를 위한 Open distory knn 알고리즘이다.

 

가볍고 효율적인 NMSLIB (Non-Metric Space Library)를 사용하여 구축된 k-NN을 사용하면 일반 Elasticsearch 쿼리를 실행하는 것과 같은 방식으로 수천 차원에서 수십억 개의 문서에 대해 대규모의 대기 시간이 가장 가까운 이웃 검색이 가능하다.

 

관리되는 검색 엔진이 효율적인 고차원 벡터 검색을 지원하면 몇 가지 이점이 있다. 유사도 검색 점수와 텍스트 및 범주 형 필드와 같은 다른 유형의 필드 점수는 점수 매기기 기능에서 병합할 수 없지만 유사도 검색에서 반환된 문서에 대한 사후 필터는 쉽게 구현된다.

 

이번 테스트를 통해 kNN 검색이 유용하다면, 실제 어플리케이션에 적용해보고 싶은 마음도 있다.

 

아래는 희망하는 요구사항이다.

• VECTOR의 Demension : 1,000 이상
• #of vectors: 1,000,000이상
• 서버 스펙 : AWS EC2 r5.large (2 Cores CPU, 16 GB Memory)
• #of servers : 1
• Latency: 10ms - 20ms

 

Data

DeepFashion과 DeepFation2를 사용. 몇몇의 겹치는 이미는 평가에 있어 크게 이슈로 작용하지 않는다. 이미지의 전체 개수는 대략 1M(991,257)이다

 

 

Feature Extraction

ImageNet으로 사전 학습된 MobileNetV2를 사용하였다. 벡터의 차원은 1,280이다.

 

필요한 가상 환경 세팅

# 가상환경 생성
conda create -n tensor python=3.6.5

# pip / python 경로 확인 (가상환경 경로인지 확인)
which pip
which python

# 필요한 라이브러리 설치
pip install tensorflow
pip install faiss-gpu cudatoolkit=10.0 -c pytorch
pip install scikit-learn

 

Example Code

import struct
import glob
import numpy as np
import tensorflow as tf
from tensorflow.keras.applications.mobilenet_v2 import preprocess_input
import tensorflow.keras.layers as layers
from tensorflow.keras.models import Model


# 이미지 전처리 프로세싱
def preprocess(img_path, input_shape):
    img = tf.io.read_file(img_path)
    img = tf.image.decode_jpeg(img, channels=input_shape[2])
    img = tf.image.resize(img, input_shape[:2])
    img = preprocess_input(img)
    return img


def main():
    batch_size = 100
    input_shape = (224, 224, 3)
    base = tf.keras.applications.MobileNetV2(input_shape=input_shape,
                                             include_top=False,
                                             weights='imagenet')
    base.trainable = False
    model = Model(inputs=base.input, outputs=layers.GlobalAveragePooling2D()(base.output))

    fnames = glob.glob('deepfashion*/**/*.jpg', recursive=True)
    list_ds = tf.data.Dataset.from_tensor_slices(fnames)
    ds = list_ds.map(lambda x: preprocess(x, input_shape), num_parallel_calls=-1)
    dataset = ds.batch(batch_size).prefetch(-1)

    with open('fvecs.bin', 'wb') as f:
        for batch in dataset:
            fvecs = model.predict(batch)
						
            #fvecs의 길이 만큼 fmt설정
            fmt = f'{np.prod(fvecs.shape)}f'

            #fmt = 포맷, fvecs.flatten() = 벡터를 한줄로 변환
            #struct.pack을 사용하여 패킹한다.
            f.write(struct.pack(fmt, *(fvecs.flatten())))

    with open('fnames.txt', 'w') as f:
        f.write('\n'.join(fnames))

if __name__ == '__main__':
    main()

 

🚩 위에서 저장되는 fvecs.bin 상세 확인

# 실제로 저장 된 fvecs.bin 로딩

with open("fvecs.bin", "rb") as f:
    fvecs = f.realines()

print(type(fvecs)) # list

print(len(fvecs)) # 3869159

print(vec[0])

> 리스트 형태로 데이터가 들어있다. 총개수는 총 데이터 수

> 안쪽 값은 struct.pack으로 패킹되어있다.

> 실제 값을 확인해보면 아래와 같이 바이너리 형태로 압축되어있는 것을 확인할 수 있다.

 

 

🚩 np.prod 사용 : 각 배열의 요소의 곱

# np.prod 

print(fvecs.shape) #(22, 1280)

print(np.prod(fvecs.shape)) #22 * 1280 = 28,160

# f-string 방식으로 포맷 형태 설정
fmt = f'{np.prod(fvecs.shape)}f'

print(fmt) #28160f

> 위에서 사용하는 f-string 방식은 파이썬 3 버전 이상에서 지원한다.

 

Prior Evaluation

유명한 유사성 검색 라이브러리 인 Faiss에는 HNSW 기능도 있습니다. 이제 성능을 보고 매개 변수를 선택한다. 아마존 엘라스틱 서비스에서 코사인 유사도가 리턴되기 때문에, L2-norm이 1이 되고, 반환된 L2 거리가 이 코드에서 코사인 유사으로 변환되도록 벡터가 정규화된다.

 

import os
import time
import math
import random
import numpy as np
import json
from sklearn.preprocessing import normalize
import faiss


def dist2sim(d):
    return 1 - d / 2
    

def get_index(index_type, dim):
    if index_type == 'hnsw':
        m = 48
        index = faiss.IndexHNSWFlat(dim, m)
        index.hnsw.efConstruction = 128
        return index
    elif index_type == 'l2':
        return faiss.IndexFlatL2(dim)
    raise


def populate(index, fvecs, batch_size=1000):
    nloop = math.ceil(fvecs.shape[0] / batch_size)
    for n in range(nloop):
        s = time.time()
        index.add(normalize(fvecs[n * batch_size : min((n + 1) * batch_size, fvecs.shape[0])]))
        print(n * batch_size, time.time() - s)

    return index


def main():
    dim = 1280
    fvec_file = 'fvecs.bin'
    index_type = 'hnsw'
    #index_type = 'l2'

	# f-string 방식 (python3 이상에서 지원)
    index_file = f'{fvec_file}.{index_type}.index'

    fvecs = np.memmap(fvec_file, dtype='float32', mode='r').view('float32').reshape(-1, dim)

    if os.path.exists(index_file):
        index = faiss.read_index(index_file)
        if index_type == 'hnsw':
            index.hnsw.efSearch = 256
    else:
        index = get_index(index_type, dim)
        index = populate(index, fvecs)
        faiss.write_index(index, index_file)
    print(index.ntotal)

    random.seed(2020)
    
	# random하게 쿼리 인덱스를 생성한다
    # 0부터 데이터 갯수 사이의 인덱스
    q_idx = [random.randint(0, fvecs.shape[0]) for _ in range(100)]

    k = 10
    s = time.time()
    
    dists, idxs = index.search(normalize(fvecs[q_idx]), k)
    print((time.time() - s) / len(q_idx))

    s = time.time()
    for i in q_idx:
        dists, idxs = index.search(normalize(fvecs[i:i+1]), k)
        
    print((time.time() - s) / len(q_idx))


if __name__ == '__main__':
    main()

 

🚩 np.memmap

import numpy as np

# 실제 코드에서 사용한 np.memmap
# 저장된 파일을 읽어옴
fvecs = np.memmap(fvec_file, dtype='float32', mode='r').view('float32').reshape(-1, dim)

# 예시
np.memmap(fvec_file, dtype='float32', mode='r').shape # (270204160, )
np.memmap(fvec_file, dtype='float32', mode='r').view('float32').reshape(-1, dim). shape # (289222, 1280)

> numpy memmap : memory-map to an array on disk

> 대용량 넘 파이를 다루다 보면 메모리가 가득 차는 경우 발생, 속도는 느리더라도 memory를 적게 쓰고 disk를 사용하는 방법

 

Search Latency

이번 테스트에서, 매우 실용적인 검색 시간을 성취했다. 비록 정확한 recall 평가는 이루어지지 않았지만, m=48, efConstruction=128, efSearch=256인 파라미터들은 실용적인 결과를 리턴했다. 이 테스트에서 , 오직 1개의 r5.large인스턴스만을 사용하였다.

 

 

Similar Image Search Result

왼쪽 칼럼은 쿼리 이미지이다. 다른 컬럼은 유사성 기준으로 정렬된 검색 결과이다. 그 결과는 흥미롭고 놀랍다. ImageNet으로 미리 학습된 모델이 fashion이미지로 최적화하지 않았음에도 결과는 꽤 실용적이다.

 

 

Amazon Elasticsearch Service

AWS 콘솔을 통해, 몇 번의 클릭으로 엘라스틱 인스턴스를 생성할 수 있다. kNN을 가능하게 하는 특별 한 설정은 없으며, 서비스로서 가능하게 하는데 10-15분 정도만 소요된다. 오직 한 개의 r5.large인스턴스를 평가를 위해 사용되었다. 

 

Data Insert 

[ 이곳부터 실행하는 내용은 아마존 인스턴스를 사용해야 따라 할 수 있다.]

1M 개의 벡터를 엘라스틱 서버로 넣는데 대략 1시간 정도 소요된다.

# 필요한 라이브러리 로딩
import time
import math
import numpy as np
import json
import certifi
from elasticsearch import Elasticsearch, helpers
from sklearn.preprocessing import normalize

# DIMENSION  수 선언
dim = 1280

# 사전에 벡터로 저장한 파일 로딩
fvecs = np.memmap('fvecs.bin', dtype='float32', mode='r').view('float32').reshape(-1, dim)

# 엘라스틱에 저장할 인덱스
idx_name = 'imsearch'

# 엘라스틱 인스턴스에 연결
es = Elasticsearch(hosts=['https://vpc-xxxxxxxxxxx.us-west-2.es.amazonaws.com'],
                   ca_certs=certifi.where())

# CLUSTER 세팅
res = es.cluster.put_settings({'persistent': {'knn.algo_param.index_thread_qty': 2}})
print(res)

# MAPPING 타입 선언
mapping = {
    'settings' : {
        'index' : {
            'knn': True,
            'knn.algo_param' : {
                'ef_search' : 256,
                'ef_construction' : 128,
                'm' : 48
            },
            'refresh_interval': -1,
            'translog.flush_threshold_size': '10gb',
            'number_of_replicas': 0
        },
    },
    'mappings': {
        'properties': {
            'fvec': {
                'type': 'knn_vector',
                'dimension': dim
            }
        }
    }
}

# 인덱스 생성
res = es.indices.create(index=idx_name, body=mapping, ignore=400)
print(res)

# 배치 사이즈
bs = 200

# 총 반복 횟수
nloop = math.ceil(fvecs.shape[0] / bs)

# 데이터 적재 진행
# 벌크로 데이터를 넣는다.
for k in range(nloop):
    rows = [{'_index': idx_name, '_id': f'{i}',
             '_source': {'fvec': normalize(fvecs[i:i+1])[0].tolist()}}
             for i in range(k * bs, min((k + 1) * bs, fvecs.shape[0]))]
    s = time.time()
    helpers.bulk(es, rows, request_timeout=30)
    print(k, time.time() - s)

 

Search

# 상위 K개 선언
k = 10

# ELASTIC 검색문
res = es.search(request_timeout=30, 
                index=idx_name,
                body={'size': k, 
                         '_source': False,
                            'query': {
                            'knn': {
                              'fvec': {
                              'vector': normalize(fvecs[0:1])[0].tolist(), 
                              'k': k
                              }
                            }
                          }
                      })
print(json.dumps(res, indent=2))

> 이전의 평가와 비교하면, 리턴 받은 Document와 image는 동일하다.

> 그러나 대략 하나의 쿼리 요청 당 걸리는 지연 시간이 15s 정도로, 이전과 비교해 너무 느린 결과를 얻었다

 

Conclusion

만약 평가가 정확하다면, 엘라스틱을 사용하여 유사 검색을 구현하는 것은 벡터 길이와 같은 것에 제한을 가져올 것으로 보인다. 이미지 검색분야에서 PCA와 같은 차원 축소를 적절하게 진행해야 한다. 다른 종류의 임베딩의 경우 제대로 작동하지 않을 수 있다.

 

elasticsearch의 유사성 검색 기능이 곧 개선되고 실용화되기를 바란다. 고차원 벡터 필드를 포함하여 수십억 개의 스케일 데이터가 서비스에서 올바르게 처리될 수 있다면 많은 실제 응용 프로그램이 이를 이용할 것입니다.