python - example - Reindexing Búsqueda elástica a través de Bulk API, escaneo y desplazamiento

python elasticsearch delete index

Question

Estoy intentando volver a indexar mi configuración de búsqueda elástica, actualmente estoy mirando la documentación de búsqueda Elastic y un ejemplo con la API de Python

Sin embargo, estoy un poco confundido sobre cómo funciona todo esto. Pude obtener la ID de espiral de la API de Python:

es = Elasticsearch("myhost")

index = "myindex"
query = {"query":{"match_all":{}}}
response = es.search(index= index, doc_type= "my-doc-type", body= query, search_type= "scan", scroll= "10m")

scroll_id = response["_scroll_id"]

Ahora mi pregunta es, ¿de qué sirve esto para mí? ¿Qué me da conocer el ID de desplazamiento? La documentación dice que use la "API a granel", pero no tengo idea de cómo los factores scoll_id en esto, fue un poco confuso.

¿Alguien podría dar un breve ejemplo que muestre cómo volver a indexar desde este punto, teniendo en cuenta que tengo el scroll_id correctamente?

Answer 1

Hola, puedes usar scroll api para recorrer todos los documentos de la forma más eficiente. Usando scroll_id puede encontrar una sesión que está almacenada en el servidor para su solicitud de desplazamiento específica. Por lo tanto, debe proporcionar scroll_id con cada solicitud para obtener más elementos.

La API a granel es para documentos de indexación más eficientes. Cuando copia e indexa necesita ambos, pero en realidad no están relacionados.

Tengo un código de Java que podría ayudarte a obtener una mejor idea sobre cómo funciona.

    public void reIndex() {
    logger.info("Start creating a new index based on the old index.");

    SearchResponse searchResponse = client.prepareSearch(MUSIC_INDEX)
            .setQuery(matchAllQuery())
            .setSearchType(SearchType.SCAN)
            .setScroll(createScrollTimeoutValue())
            .setSize(SCROLL_SIZE).execute().actionGet();

    BulkProcessor bulkProcessor = BulkProcessor.builder(client,
            createLoggingBulkProcessorListener()).setBulkActions(BULK_ACTIONS_THRESHOLD)
            .setConcurrentRequests(BULK_CONCURRENT_REQUESTS)
            .setFlushInterval(createFlushIntervalTime())
            .build();

    while (true) {
        searchResponse = client.prepareSearchScroll(searchResponse.getScrollId())
                .setScroll(createScrollTimeoutValue()).execute().actionGet();

        if (searchResponse.getHits().getHits().length == 0) {
            logger.info("Closing the bulk processor");
            bulkProcessor.close();
            break; //Break condition: No hits are returned
        }

        for (SearchHit hit : searchResponse.getHits()) {
            IndexRequest request = new IndexRequest(MUSIC_INDEX_NEW, hit.type(), hit.id());
            request.source(hit.sourceRef());
            bulkProcessor.add(request);
        }
    }
}

Answer 2

Para cualquiera que se encuentre con este problema, puede usar la siguiente API del cliente de Python para reindexar:

https://elasticsearch-py.readthedocs.org/en/master/helpers.html#elasticsearch.helpers.reindex

Esto le ayudaría a evitar tener que desplazarse y buscar para obtener todos los datos y usar la API masiva para colocar datos en el nuevo índice.

Answer 3

aquí hay un ejemplo de reindexación a otro nodo elasticsearch usando elasticsearch-py:

from elasticsearch import helpers
es_src = Elasticsearch(["host"])
es_des = Elasticsearch(["host"])

helpers.reindex(es_src, ''src_index_name'', ''des_index_name'', target_client=es_des)

también puede reindexar el resultado de una consulta en un índice diferente, aquí está cómo hacerlo:

from elasticsearch import helpers
es_src = Elasticsearch(["host"])
es_des = Elasticsearch(["host"])

body = {"query": {"term": {"year": "2004"}}}
helpers.reindex(es_src, ''src_index_name'', ''des_index_name'', target_client=es_des, query=body)