Elasticsearch java API bulk delete not working

Question

Я пытаюсь выполнить массовое удаление документов, чьи идентификаторы получены из предыдущего поиска. Запрос для определения документов, которые являются кандидатами на удаление, дает желаемые результаты (тысячи записей), однако массовое удаление удаляет только 10 записей за раз, даже если я загружаю все результаты исходного запроса;

Client client = node.client(); 
BulkRequestBuilder bulkRequest = client.prepareBulk(); 

SearchResponse deletes = client.prepareSearch("my_index") 
     .setTypes("my_doc_type") 
     .setSearchType(SearchType.DFS_QUERY_THEN_FETCH) 
     .setQuery(boolQuery().mustNot(termQuery("tId", transactionId))) 
     .execute() 
     .actionGet(); 

long deleteHits = deletes.getHits().getTotalHits(); 

if (deleteHits > 0) { 

    logger.info("Preparing to delete (" + deleteHits + ") " + 
      "documents from index"); 

    Arrays.asList(deletes.getHits().getHits()).stream().forEach(h -> 
      bulkRequest.add(client.prepareDelete() 
       .setIndex("my_index") 
       .setType("my_doc_type") 
       .setId(h.getId()))); 
    } 

    BulkResponse bulkResponse = bulkRequest.execute().actionGet(); 

    if (bulkResponse.hasFailures()) { 
     throw new RuntimeException(bulkResponse.buildFailureMessage()); 
    } 

} 

Answer 1

По умолчанию ответа поиска возвращает только 10 лучших результатов. Таким образом, хотя deletes .getHits().getTotalHits() может быть в тысячах или даже миллионах, размер deletes.getHits().getHits() никогда не будет больше, чем вы указали в параметре size вашего запроса, который по умолчанию 10.

Наивным подходом было бы попытаться выполнить разбиение на страницы с помощью обычного поиска, изменив параметр from. Однако это может привести к отсутствию удаления некоторых записей, так как каждая команда будет выполнять новый поиск, а результат следующего поиска может сдвинуться по сравнению с предыдущим поиском в результате удаления записей в предыдущем поиске.

Правильный подход заключается в использовании специализированных scan and scroll search для разбивки страниц на страницы. Этот тип поиска будет поддерживать согласованные результаты между вызовами. Пример этого подхода можно найти в плагине delete by query, который будет доступен в версии 2.0.

Также нужно отметить, что в то время как delete by query функциональность существует в предыдущих версиях elasticsearch и это может показаться самым простым решением для вашей проблемы, я бы рекомендовал использовать сканирование/свитку из poor performance and fragility существующих удалить, реализация API запросов в pre-v2.0.

Answer 2

deletes.getHits().getTotalHits дать вам общее количество обращений для поиска, но SearchResponse deletes не содержит все результаты. Вам нужно будет разбивать на части.

вам нужно использовать что-то вроде этого, чтобы определить пейджинговый

client.prepareSearch("my_index").setFrom(int from).setSize(int pageSize);