jjeong

arirang plug 을 적용해서 돌려본 실행 속도 입니다.

[Settings]

"settings":{
  "number_of_shards":1,
  "number_of_replicas":0,
  "index.refresh_interval":"1h",
  "index.routing.allocation.require.box_type":"indexing",
  "index.similarity.default.type":"BM25",
  "index":{
    "analysis":{
      "analyzer":{
        "arirang_custom_analyzer":{
          "tokenizer":"arirang_tokenizer",
          "filter":[
              "lowercase",
              "trim",
              "custom_synonym",
              "arirang_filter"
            ]
          }
      },
      "filter":{
        "custom_synonym":{
            "type":"synonym_graph",
            "synonyms":[]
          }
        }
      }
  }
}

[Mappings]

"mappings" : {
  "_doc": {
    "_source": {
      "enabled": true
    },
    "dynamic_templates": [
      {
        "strings": {
          "match_mapping_type": "string",
          "mapping": {
            "type": "text",
            "analyzer": "arirang_custom_analyzer",
              "fields": {
                "raw": {
                  "type":  "keyword",
                  "ignore_above": 50
                }
            }
          }
        }
      }
    ]
  }
}

얼마전 밋업에서 김종민님이 공유해 주신 위키피디아 문서를 가지고 테스트한 결과 입니다.

그냥 제가 만든 Indexer 모듈에 대한 성능 테스트용도로 한번 실험 삼아 돌려 본거라 별 도움이 안될 수 있습니다.

아래 문서는 언제까지 올라가 있을지 모르겠습니다.

빨리 받으시면 될 듯 하내요.

문서 다운로드 링크)

https://s3.ap-northeast-2.amazonaws.com/kr.elastic.co/sample-data/kr-wikipedia-dump.json.tar.gz

위키피디아 문서 색인 테스트 결과)

결과적으로 튜닝한 내용은 전혀 없고 그냥 색인기 기능 테스트 정도로 보셔도 될 것 같습니다.

_all 필드는 앞으로 사라질 예정입니다.

그러니 사용하지 않도록 합니다.

기본 _all 필드는 disabled 이기 때문에 별도 mapping 설정에 추가 하지 않으셔도 됩니다.

만약 사용 하실 거라면 아래 문서를 참고하세요.

Reference)

https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-all-field.html

_all enable 시 아래와 같은 경고 문구를 보실 수 있습니다.

이번에 새로 Indexer 모듈을 만들다 보니 아래 문서의 내용을 확인하게 되었습니다.

Reference)

https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/client.html

https://github.com/elastic/elasticsearch/tree/master/client/rest/src/test/java/org/elasticsearch/client

원문발췌)

We plan on deprecating the TransportClient in Elasticsearch 7.0 and removing it completely in 8.0. Instead, you should be using the Java High Level REST Client, which executes HTTP requests rather than serialized Java requests. The migration guidedescribes all the steps needed to migrate.

The Java High Level REST Client currently has support for the more commonly used APIs, but there are a lot more that still need to be added. You can help us prioritise by telling us which missing APIs you need for your application by adding a comment to this issue: Java high-level REST client completeness.

Any missing APIs can always be implemented today by using the low level Java REST Client with JSON request and response bodies.

사실 저도 TransportClient 를 사용하면 version 때문에 고민을 하긴했었는데요.

이 참에 그냥 REST Client 로 넘어 가야 겠다고 생각해서 새로 작성 하였습니다.

기본적인 예제 코드는 아래 내용 참고 하시면 됩니다.

    public static void main(String[] args) {
        requestRestClient();
    }

    public static void requestRestClient() {
        int numNodes = 5; //randomIntBetween(1, 5);
        HttpHost[] hosts = new HttpHost[numNodes];
        for (int i = 0; i < numNodes; i++) {
            hosts[i] = new HttpHost("localhost", 9200);
        }
        RestClientBuilder builder = RestClient.builder(hosts);
        RestClient restClient = builder.build();

        builder.setMaxRetryTimeoutMillis(10000);

        builder.setFailureListener(new RestClient.FailureListener() {
            @Override
            public void onFailure(HttpHost host) {
                System.out.println("onFailure");
            }
        });

        builder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
            @Override
            public RequestConfig.Builder customizeRequestConfig(
                RequestConfig.Builder requestConfigBuilder) {
                return requestConfigBuilder.setSocketTimeout(10000);
            }
        });

        Map<String, String> params = new HashMap<>();
        params.put("preference", "_shards:0,1,2,3");

        HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory consumerFactory =
            new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(
                30 * 1024 * 1024);
        try {
            Response response = restClient
                .performRequest("GET", "/hello/_search", params, null, consumerFactory);

            System.out.println(response.toString());

            // Read Case 1
            RequestLine requestLine = response.getRequestLine();
            HttpHost host = response.getHost();
            int statusCode = response.getStatusLine().getStatusCode();
            Header[] headers = response.getHeaders();
            String responseBody = EntityUtils.toString(response.getEntity());

            System.out.println(requestLine);
            System.out.println(host);
            System.out.println(statusCode);
            System.out.println(headers);
            System.out.println(responseBody);

            // Read Case 2 - old fashion
            BufferedReader reader = new BufferedReader(new InputStreamReader(
                response.getEntity().getContent()));

            String inputLine;
            StringBuffer buffer = new StringBuffer();

            while ((inputLine = reader.readLine()) != null) {
                buffer.append(inputLine);
            }

            reader.close();

//            System.out.println(buffer.toString());
        } catch (IOException e) {
            e.printStackTrace();
        }

        try {
            restClient.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static int randomIntBetween(int min, int max) {
        return RandomNumbers.randomIntBetween(random(), min, max);
    }

    public static Random random() {
        return RandomizedContext.current().getRandom();
    }

Elasticsearch 6.3 이 릴리즈 되면서 가장 큰 변화는 Elastic 용과 OSS 용 이렇게 두개로 설치 패키지가 나뉜것입니다.

즉, Elastic 사의 license 를 기반으로 동작하는 x-pack 들을 사용하기 위해서는 Elastic 버전을 설치 하셔야 한다는 이야기 입니다.

Basic  라이센스의 경우 free 이기 때문에 OSS 버전을 설치 하고 사용할 수 있겠다 생각 하시면 안되는 걸 경험 하실 수 있습니다. (제가 그랬습니다 ^^;)

전에는 x-pack 을 elasticsearch-plugin 으로 설치가 가능 했지만 6.3 에서는 지원을 하고 있지 않습니다.

또한 x-pack 을 쓰기 위해서는 default system index 생성이 필요 한데 아래 설정이 되어 있으면 오류가 발생을 하게 됩니다.

이 설정은 잘 아시겠지만 index 가 없을 때 자동으로 생성을 하게 해주는 설정인데요.

저는 보안이랑 관리적인 부분에서 disable 하고 사용했었는데 basic 버전을 사용하기 위해서 이 설정을 과감히 포기 했습니다.

몇 가지 removed 된 설정들이 있는데요.

5.x 올라 오면서 변경된 내용과 일부 비슷한 것들이 있어서 링크 겁니다.

[Elasticsearch] 2.4.x to 5.2.x 으로의 elasticsearch.yml

그리고 Elastic 에서 제공하는 Breaking Changes 를 보시는 것도 도움이 됩니다.

정리하면)

6.x 에 추가된 API 중 맘에 드는 것이 있어서 글을 퍼왔습니다.

Reference)

https://www.elastic.co/guide/en/elasticsearch/reference/current/search.html

Adaptive Replica Selection

As an alternative to requests being sent to copies of the data in a round robin fashion, you may enable adaptive replica selection. This allows the coordinating node to send the request to the copy deemed "best" based on a number of criteria:

• Response time of past requests between the coordinating node and the node containing the copy of the data
• Time past search requests took to execute on the node containing the data
• The queue size of the search threadpool on the node containing the data

This can be turned on by changing the dynamic cluster setting cluster.routing.use_adaptive_replica_selection from false to true:

위에 기술 되어 있는 것 처럼,

- 기본적으로는 Round Robin 방식으로 동작 합니다.

- 하지만 3가지 기준을 가지고 이 기능은 동작을 합니다.

이 중에서 가장 좋은 기준을 바탕으로 동작 하게 되는 것입니다.

평소 Get API 에 대해서 설명을 할 때 document id 로 lookup 을 하기 때문에 모든 shard 로 request 를 보내지 않아 빠르다고 설명을 했습니다.

이 과정에서 놓치기 쉬운 부분이 있는데 기본적으로 Get API 는 realtime  동작을 합니다.

즉, refresh 동작과 상관 없이 기본적으로 최신 정보를 가져오게 되는 것입니다.

다시 말해, 내부적으로 refresh 동작을 수행한다는 의미가 됩니다.

그래서 Get API 에서는 realtime 이라는 변수를 통해서 realtime Get 을 사용 할 것인지 말 것인지를 정의 할 수 있습니다.

아래는 위 설명을 이해 하는데 도움을 주기 위해서 소스 코드의 일부를 발췌 하였습니다.

[GetRequest.java]

private boolean refresh = false;

boolean realtime = true;

/**
 * Should a refresh be executed before this get operation causing the operation to
 * return the latest value. Note, heavy get should not set this to {@code true}. Defaults
 * to {@code false}.
 */
public GetRequest refresh(boolean refresh) {
    this.refresh = refresh;
    return this;
}

public boolean refresh() {
    return this.refresh;
}

public boolean realtime() {
    return this.realtime;
}

@Override
public GetRequest realtime(boolean realtime) {
    this.realtime = realtime;
    return this;
}

[RestGetAction.java]

@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    final boolean includeTypeName = request.paramAsBoolean("include_type_name", true);
    final String type = request.param("type");
    if (includeTypeName == false && MapperService.SINGLE_MAPPING_NAME.equals(type) == false) {
        throw new IllegalArgumentException("You may only use the [include_type_name=false] option with the get APIs with the " +
                "[{index}/_doc/{id}] endpoint.");
    }
    final GetRequest getRequest = new GetRequest(request.param("index"), type, request.param("id"));
    getRequest.refresh(request.paramAsBoolean("refresh", getRequest.refresh()));
    getRequest.routing(request.param("routing"));
    getRequest.preference(request.param("preference"));
    getRequest.realtime(request.paramAsBoolean("realtime", getRequest.realtime()));
    if (request.param("fields") != null) {
        throw new IllegalArgumentException("the parameter [fields] is no longer supported, " +
            "please use [stored_fields] to retrieve stored fields or [_source] to load the field from _source");
    }
    final String fieldsParam = request.param("stored_fields");
    if (fieldsParam != null) {
        final String[] fields = Strings.splitStringByCommaToArray(fieldsParam);
        if (fields != null) {
            getRequest.storedFields(fields);
        }
    }

    getRequest.version(RestActions.parseVersion(request));
    getRequest.versionType(VersionType.fromString(request.param("version_type"), getRequest.versionType()));

    getRequest.fetchSourceContext(FetchSourceContext.parseFromRestRequest(request));

    return channel -> client.get(getRequest, new RestToXContentListener<GetResponse>(channel) {
        @Override
        protected RestStatus getStatus(final GetResponse response) {
            return response.isExists() ? OK : NOT_FOUND;
        }
    });
}

[TransportGetAction.java]

@Override
protected void asyncShardOperation(GetRequest request, ShardId shardId, ActionListener<GetResponse> listener) throws IOException {
    IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
    IndexShard indexShard = indexService.getShard(shardId.id());
    if (request.realtime()) { // we are not tied to a refresh cycle here anyway
        listener.onResponse(shardOperation(request, shardId));
    } else {
        indexShard.awaitShardSearchActive(b -> {
            try {
                super.asyncShardOperation(request, shardId, listener);
            } catch (Exception ex) {
                listener.onFailure(ex);
            }
        });
    }
}

@Override
protected GetResponse shardOperation(GetRequest request, ShardId shardId) {
    IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
    IndexShard indexShard = indexService.getShard(shardId.id());

    if (request.refresh() && !request.realtime()) {
        indexShard.refresh("refresh_flag_get");
    }

    GetResult result = indexShard.getService().get(request.type(), request.id(), request.storedFields(),
            request.realtime(), request.version(), request.versionType(), request.fetchSourceContext());
    return new GetResponse(result);
}

보시면 아시겠지만 refresh 와 realtime 의 동작에는 차이가 있습니다.

이해 하기 쉽게 정리 하면,

- refresh 는 get 수행 전에 relevant shard 를 대상으로 refresh 동작을 먼저 하기 때문에 성능 저하가 있을 수 있습니다.

- realtime 은 수행 시점에 refresh 동작을 수행 하게 됩니다. (이 경우 refresh searcher 의 대상은 internal searcher 가 됩니다.)

- 여기서 translog 에서 데이터를 읽어야 하는 경우가 있는데 이 경우는 update API 에서 사용 되게 됩니다.

[InternalEngine.java]

@Override
public GetResult get(Get get, BiFunction<String, SearcherScope, Searcher> searcherFactory) throws EngineException {
    assert Objects.equals(get.uid().field(), IdFieldMapper.NAME) : get.uid().field();
    try (ReleasableLock ignored = readLock.acquire()) {
        ensureOpen();
        SearcherScope scope;
        if (get.realtime()) {
            VersionValue versionValue = null;
            try (Releasable ignore = versionMap.acquireLock(get.uid().bytes())) {
                // we need to lock here to access the version map to do this truly in RT
                versionValue = getVersionFromMap(get.uid().bytes());
            }
            if (versionValue != null) {
                if (versionValue.isDelete()) {
                    return GetResult.NOT_EXISTS;
                }
                if (get.versionType().isVersionConflictForReads(versionValue.version, get.version())) {
                    throw new VersionConflictEngineException(shardId, get.type(), get.id(),
                        get.versionType().explainConflictForReads(versionValue.version, get.version()));
                }
                if (get.isReadFromTranslog()) {
                    // this is only used for updates - API _GET calls will always read form a reader for consistency
                    // the update call doesn't need the consistency since it's source only + _parent but parent can go away in 7.0
                    if (versionValue.getLocation() != null) {
                        try {
                            Translog.Operation operation = translog.readOperation(versionValue.getLocation());
                            if (operation != null) {
                                // in the case of a already pruned translog generation we might get null here - yet very unlikely
                                TranslogLeafReader reader = new TranslogLeafReader((Translog.Index) operation, engineConfig
                                    .getIndexSettings().getIndexVersionCreated());
                                return new GetResult(new Searcher("realtime_get", new IndexSearcher(reader)),
                                    new VersionsAndSeqNoResolver.DocIdAndVersion(0, ((Translog.Index) operation).version(), reader, 0));
                            }
                        } catch (IOException e) {
                            maybeFailEngine("realtime_get", e); // lets check if the translog has failed with a tragic event
                            throw new EngineException(shardId, "failed to read operation from translog", e);
                        }
                    } else {
                        trackTranslogLocation.set(true);
                    }
                }
                refresh("realtime_get", SearcherScope.INTERNAL);
            }
            scope = SearcherScope.INTERNAL;
        } else {
            // we expose what has been externally expose in a point in time snapshot via an explicit refresh
            scope = SearcherScope.EXTERNAL;
        }

        // no version, get the version from the index, we know that we refresh on flush
        return getFromSearcher(get, searcherFactory, scope);
    }
}

[코드의 이해]

이미 synonym filter 테스트 관련 글을 공유 했었습니다.

- [Elasticsearch] Synonym filter 테스트

이 테스트에서 발생한 문제는 동의어에 대한 position 정보가 잘 못되는 것입니다.

테스트 환경은 기본 Elasticsearch 6.x 에서 진행 되었습니다. 

아래는 SynonymFilter 코드 내 주석 입니다.

[SynonymFilter.java]

그리고 아래는 동의어 처리 시 문제가 발생 하는 부분의 코드 입니다.

[SynonymMap.java]

기본적으로 동의어 처리에 대한 문제는 이미 lucene 레벨에서 개선이 되었습니다.

관련 참고 링크는 아래와 같습니다.

[Reference links]

https://issues.apache.org/jira/browse/LUCENE-6664

http://blog.mikemccandless.com/2012/04/lucenes-tokenstreams-are-actually.html

이와 같이 개선된 synonym filter 를 elasticsearch 에서는 아래와 같이 사용 할 수 있습니다.

[Reference links]

https://www.elastic.co/guide/en/elasticsearch/reference/master/analysis-synonym-graph-tokenfilter.html

[Create index]

[Request analyze]

[Analyzed result]

특별히 코드를 수정 하거나 하지 않고 문제가 해결 된 것을 확인 하실 수 있습니다.

왜 해결 되었는지는 위 synonym graph filter 에 대해서 문서를 보시면 되겠습니다.

Elastic 사의 공식 문서를 보시면 쉽게 하실 수 있습니다.

기본 flow 는 아래와 같습니다.

각각에 필요한 참조문서는

[Logstash Input File]

https://www.elastic.co/guide/en/logstash/current/plugins-inputs-file.html

[Logstash Filter CSV]

https://www.elastic.co/guide/en/logstash/current/plugins-filters-csv.html

[Logstash Output Elasticsearch]

https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html

[Elasticsearch Indices Templates]

https://www.elastic.co/guide/en/elasticsearch/reference/master/indices-templates.html

Template 이 필요한 이유는 csv 파일 데이터에 대한 dynamic mapping 시 의도치 않은 데이터에 대한 형변환 오류를 방지 하기 위함 입니다.

사전에 꼭 정의 하셔서 reindexing 하는 일이 없도록 주의 하시면 좋을 것 같습니다.

커뮤니티에 질문 주신 내용이 있어서 바쁘지만 테스트 결과 공유 드립니다.

커뮤니티에 질문 올려 주신 분이 계셔서 직접 테스트 진행 했습니다.

제가 position 관련 에러 수정을 위한 테스트 시간이 없어서 인명 사전 등록 방법으로 처리 했는데요.

수정이 필요 하시면 KoreanFilter 코드를 수정 하시면 됩니다.

설치 및 테스트 Elasticsearch Version)

Arirang plugin 설치)

Index 삭제)

Index 생성)

Analyze 실행)

동의어 처리가 되지 않은 이유)

tokenizer 에서 추출한 토큰은 위와 같이 두개 입니다.

아래는 filter 를 적용한 내용입니다.

위에 추출된 term 목록을 보면 "마왕" 이라는 동의어가 추가 되지 않은것을 볼 수 있습니다.

이것은 두 가지 방법으로 해결이 가능 합니다.

사전에 "신해철" 이라는 인명 사전 정보를 등록 하시면 됩니다.

기본적으로 tokenzier 과정 후 filter 처리가 되면서 사전에 등록된 정보로 term 구성이 되기 때문에 사전에 누락 된 경우는 일반적으로 KoreanFilter 에 의해서 bigram 처리가 됩니다.

다른 한 가지는 position 정보가 구성 시 오류가 나지 않도록 코드를 수정 하는 것입니다.

KoreanFilter 코드를 참고 하셔서 테스트 및 수정 하시면 됩니다.

아래는 사전에 "신해철" 추가 후 실행한 방법 입니다.

Elasticsearch docker 만들어 보기를 참고하세요.

초간단 예제를 보여 드리겠습니다.

Dockerfile 에 추가해 주시면 됩니다.

참 쉽습니다.