Use the Elasticsearch client RestHighLevelClient to connect to LindormSearch - Lindorm

El cliente REST de alto nivel para Java es un cliente de Elasticsearch que proporciona una API de nivel superior para la gestión de índices y documentos. LindormSearch es compatible con Elasticsearch 7.10 y versiones anteriores, por lo que puede utilizar este cliente para conectarse a LindormSearch y ejecutar consultas y búsquedas complejas sin modificar el código existente de su aplicación.

El cliente REST de alto nivel para Java ofrece compatibilidad hacia adelante; por ejemplo, la versión 6.7.0 puede comunicarse con clústeres de Elasticsearch 6.7.0 o posteriores. Utilice la versión 7.10.0 o anterior para conectarse a LindormSearch.

Requisitos previos

Antes de empezar, asegúrese de cumplir con lo siguiente:

Tener instalado JDK 1.8 o posterior
Tener habilitado el motor de búsqueda LindormSearch. Consulte Enablement Guide.
Haber agregado la dirección IP de su cliente a la lista blanca de la instancia de Lindorm. Consulte Configure whitelists.

Agregar dependencias

Para un proyecto Maven, agregue las siguientes dependencias a su archivo pom.xml:

<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>7.10.0</version>
</dependency>
<dependency>
    <groupId>org.apache.logging.log4j</groupId>
    <artifactId>log4j-core</artifactId>
    <version>2.20.0</version>
</dependency>
<dependency>
    <groupId>org.apache.logging.log4j</groupId>
    <artifactId>log4j-api</artifactId>
    <version>2.20.0</version>
</dependency>

Conectar a LindormSearch

Utilice RestClient.builder() para crear un objeto RestHighLevelClient. El cliente usa BasicCredentialsProvider para la autenticación.

// Set the Elasticsearch-compatible endpoint and port for LindormSearch.
String search_url = "ld-t4n5668xk31ui****-proxy-search-public.lindorm.rds.aliyuncs.com";
int search_port = 30070;

// Set the username and password. Retrieve them from the Lindorm console:
// navigate to Database Connections > Search Engine tab.
String username = "user";
String password = "test";

final CredentialsProvider credentials_provider = new BasicCredentialsProvider();
credentials_provider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));

RestHighLevelClient highClient = new RestHighLevelClient(
    RestClient.builder(new HttpHost(search_url, search_port, "http"))
        .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                return httpClientBuilder.setDefaultCredentialsProvider(credentials_provider);
            }
        })
);

Parámetros de conexión

Parámetro	Descripción
`search_url`	El endpoint compatible con Elasticsearch para el motor de búsqueda LindormSearch. Para obtener el endpoint, consulte Elasticsearch-compatible address. Utilice la dirección de Virtual Private Cloud (VPC) al conectarse desde una VPC, o la dirección de Internet al conectarse a través de la red pública.
`search_port`	El puerto para compatibilidad con Elasticsearch en LindormSearch: `30070`.
`username`	El nombre de usuario para el motor de búsqueda.
`password`	La contraseña para el motor de búsqueda.

Elegir un tipo de conexión de red

VPC (recomendado): Si su aplicación se ejecuta en una instancia de Elastic Compute Service (ECS) dentro de la misma VPC que la instancia de Lindorm, conéctese mediante VPC para obtener menor latencia y mayor seguridad. Establezca search_url con la dirección VPC del endpoint compatible con Elasticsearch.
Red pública: Si su aplicación se ejecuta fuera de Alibaba Cloud, primero debe habilitar el endpoint público. En la consola de Lindorm, vaya a Database Connections, haga clic en la pestaña Search Engine y luego haga clic en Enable Public Endpoint en la esquina superior derecha. Después, establezca search_url con la dirección de Internet del endpoint compatible con Elasticsearch.

Crear un índice

Utilice CreateIndexRequest para crear un índice. El siguiente ejemplo crea un índice llamado lindorm_index con 4 shards.

String index_name = "lindorm_index";

// Create a CreateIndexRequest and configure index settings.
CreateIndexRequest createIndexRequest = new CreateIndexRequest(index_name);
Map<String, Object> settingsMap = new HashMap<>();
settingsMap.put("index.number_of_shards", 4);
createIndexRequest.settings(settingsMap);

CreateIndexResponse createIndexResponse = highClient.indices().create(createIndexRequest, COMMON_OPTIONS);
if (createIndexResponse.isAcknowledged()) {
    System.out.println("Create index [" + index_name + "] successfully.");
}

Indexar un documento

Utilice IndexRequest para escribir un solo documento. Especifique un ID de documento o déjelo en blanco para que el sistema genere uno automáticamente (omitir el ID puede mejorar el rendimiento de escritura).

// Specify the document ID.
String doc_id = "test";

// Build the document fields. Replace with actual field names and values for your use case.
Map<String, Object> jsonMap = new HashMap<>();
jsonMap.put("field1", "value1");
jsonMap.put("field2", "value2");

IndexRequest indexRequest = new IndexRequest(index_name);
indexRequest.id(doc_id).source(jsonMap);

IndexResponse indexResponse = highClient.index(indexRequest, COMMON_OPTIONS);
System.out.println("Index document with id[" + indexResponse.getId() + "] successfully.");

Indexar documentos en bloque

Utilice BulkProcessor junto con bulkAsync() para escribir grandes volúmenes de documentos de manera eficiente. BulkProcessor agrupa las solicitudes de indexación individuales y las envía en bloque cuando se alcanza cualquiera de los umbrales configurados.

Configuración de BulkProcessor

Parámetro	Descripción	Valor de ejemplo
`setConcurrentRequests`	Número máximo de solicitudes en bloque simultáneas. Aumente este valor para mejorar el rendimiento bajo cargas de escritura intensas. Valor predeterminado: `1`.	`10`
`setFlushInterval`	Intervalo de tiempo tras el cual se envía una solicitud en bloque independientemente del tamaño o la cantidad. Úselo como medida de seguridad para evitar que los datos permanezcan en el búfer demasiado tiempo.	`5` segundos
`setBulkActions`	Cantidad de operaciones individuales que desencadena un vaciado. Ajuste este valor según el tamaño promedio de sus documentos.	`5000`
`setBulkSize`	Tamaño total de las operaciones en el búfer que desencadena un vaciado.	`5` MB

int bulkTotal = 100000;
AtomicLong failedBulkItemCount = new AtomicLong();

BulkProcessor.Builder builder = BulkProcessor.builder(
    (request, bulkListener) -> highClient.bulkAsync(request, COMMON_OPTIONS, bulkListener),
    new BulkProcessor.Listener() {
        @Override
        public void beforeBulk(long executionId, BulkRequest request) {}

        @Override
        public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
            // Count failed items in the bulk response.
            for (BulkItemResponse bulkItemResponse : response) {
                if (bulkItemResponse.isFailed()) {
                    failedBulkItemCount.incrementAndGet();
                }
            }
        }

        @Override
        public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
            // If this callback fires, all requests in the bulk were not executed.
            if (null != failure) {
                failedBulkItemCount.addAndGet(request.numberOfActions());
            }
        }
    });

// Maximum concurrent bulk requests. Default is 1; increase for higher throughput.
builder.setConcurrentRequests(10);
// Flush thresholds — a bulk request is sent when any of these is met.
builder.setFlushInterval(TimeValue.timeValueSeconds(5));  // every 5 seconds
builder.setBulkActions(5000);                             // every 5,000 operations
builder.setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB)); // every 5 MB

BulkProcessor bulkProcessor = builder.build();
Random random = new Random();
for (int i = 0; i < bulkTotal; i++) {
    // Replace with actual field names and values for your use case.
    Map<String, Object> map = new HashMap<>();
    map.put("field1", random.nextInt() + "");
    map.put("field2", random.nextInt() + "");
    IndexRequest bulkItemRequest = new IndexRequest(index_name);
    bulkItemRequest.source(map);
    bulkProcessor.add(bulkItemRequest);
}

// Wait up to 120 seconds for all pending operations to complete.
bulkProcessor.awaitClose(120, TimeUnit.SECONDS);
long failure = failedBulkItemCount.get(),
     success = bulkTotal - failure;
System.out.println("Bulk using BulkProcessor finished with [" + success + "] requests succeeded, [" + failure + "] requests failed.");

Buscar documentos

Envíe primero una solicitud de actualización para hacer visibles los datos escritos recientemente y luego ejecute sus consultas.

De forma predeterminada, una consulta de búsqueda devuelve un máximo de 10,000 documentos. Para obtener el recuento total exacto cuando hay más de 10,000 documentos coincidentes, llame a searchSourceBuilder.trackTotalHits(true) antes de ejecutar la consulta.

// Refresh the index to make written data searchable.
RefreshRequest refreshRequest = new RefreshRequest(index_name);
highClient.indices().refresh(refreshRequest, COMMON_OPTIONS);
System.out.println("Refresh on index [" + index_name + "] successfully.");

// Query all documents.
SearchRequest searchRequest = new SearchRequest(index_name);
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
// Uncomment the next line to get the exact total count when results exceed 10,000.
// searchSourceBuilder.trackTotalHits(true);
QueryBuilder queryMatchAllBuilder = new MatchAllQueryBuilder();
searchSourceBuilder.query(queryMatchAllBuilder);
searchRequest.source(searchSourceBuilder);
SearchResponse searchResponse = highClient.search(searchRequest, COMMON_OPTIONS);
long totalHit = searchResponse.getHits().getTotalHits().value;
System.out.println("Search query match all hits [" + totalHit + "] in total.");

// Query documents by ID.
QueryBuilder queryByIdBuilder = new MatchQueryBuilder("_id", doc_id);
searchSourceBuilder.query(queryByIdBuilder);
searchRequest.source(searchSourceBuilder);
searchResponse = highClient.search(searchRequest, COMMON_OPTIONS);
for (SearchHit searchHit : searchResponse.getHits()) {
    System.out.println("Search query by id response [" + searchHit.getSourceAsString() + "]");
}

Eliminar documentos e índices

Utilice DeleteRequest para eliminar un solo documento y DeleteIndexRequest para borrar un índice.

// Delete a single document by ID.
DeleteRequest deleteRequest = new DeleteRequest(index_name);
deleteRequest.id(doc_id);
DeleteResponse deleteResponse = highClient.delete(deleteRequest, COMMON_OPTIONS);
System.out.println("Delete document with id [" + deleteResponse.getId() + "] successfully.");

// Delete the index.
DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(index_name);
AcknowledgedResponse deleteIndexResponse = highClient.indices().delete(deleteIndexRequest, COMMON_OPTIONS);
if (deleteIndexResponse.isAcknowledged()) {
    System.out.println("Delete index [" + index_name + "] successfully.");
}

highClient.close();

Ejemplo completo

import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.HttpAsyncResponseConsumerFactory;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.MatchAllQueryBuilder;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public class RestHClientTest {
    private static final RequestOptions COMMON_OPTIONS;
    static {
        // Set the maximum response buffer size to 30 MB. The default is 100 MB.
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
        builder.setHttpAsyncResponseConsumerFactory(
            new HttpAsyncResponseConsumerFactory
                .HeapBufferedResponseConsumerFactory(30 * 1024 * 1024));
        COMMON_OPTIONS = builder.build();
    }

    public static void main(String[] args) {
        // Set the Elasticsearch-compatible endpoint and port for LindormSearch.
        String search_url = "ld-t4n5668xk31ui****-proxy-search-public.lindorm.rds.aliyuncs.com";
        int search_port = 30070;

        // Set the username and password. Retrieve them from the Lindorm console.
        String username = "user";
        String password = "test";

        final CredentialsProvider credentials_provider = new BasicCredentialsProvider();
        credentials_provider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
        RestHighLevelClient highClient = new RestHighLevelClient(
            RestClient.builder(new HttpHost(search_url, search_port, "http"))
                .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                        return httpClientBuilder.setDefaultCredentialsProvider(credentials_provider);
                    }
                })
        );

        try {
            String index_name = "lindorm_index";

            // Create an index.
            CreateIndexRequest createIndexRequest = new CreateIndexRequest(index_name);
            Map<String, Object> settingsMap = new HashMap<>();
            settingsMap.put("index.number_of_shards", 4);
            createIndexRequest.settings(settingsMap);
            CreateIndexResponse createIndexResponse = highClient.indices().create(createIndexRequest, COMMON_OPTIONS);
            if (createIndexResponse.isAcknowledged()) {
                System.out.println("Create index [" + index_name + "] successfully.");
            }

            // Index a single document.
            // Specify the document ID. If you do not specify the document ID, an ID is automatically generated,
            // which can improve write performance.
            String doc_id = "test";
            Map<String, Object> jsonMap = new HashMap<>();
            jsonMap.put("field1", "value1");
            jsonMap.put("field2", "value2");
            IndexRequest indexRequest = new IndexRequest(index_name);
            indexRequest.id(doc_id).source(jsonMap);
            IndexResponse indexResponse = highClient.index(indexRequest, COMMON_OPTIONS);
            System.out.println("Index document with id[" + indexResponse.getId() + "] successfully.");

            // Bulk index documents using BulkProcessor.
            int bulkTotal = 100000;
            AtomicLong failedBulkItemCount = new AtomicLong();
            BulkProcessor.Builder builder = BulkProcessor.builder(
                (request, bulkListener) -> highClient.bulkAsync(request, COMMON_OPTIONS, bulkListener),
                new BulkProcessor.Listener() {
                    @Override
                    public void beforeBulk(long executionId, BulkRequest request) {}

                    @Override
                    public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                        for (BulkItemResponse bulkItemResponse : response) {
                            if (bulkItemResponse.isFailed()) {
                                failedBulkItemCount.incrementAndGet();
                            }
                        }
                    }

                    @Override
                    public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                        if (null != failure) {
                            failedBulkItemCount.addAndGet(request.numberOfActions());
                        }
                    }
                });
            builder.setConcurrentRequests(10);
            builder.setFlushInterval(TimeValue.timeValueSeconds(5));
            builder.setBulkActions(5000);
            builder.setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB));
            BulkProcessor bulkProcessor = builder.build();
            Random random = new Random();
            for (int i = 0; i < bulkTotal; i++) {
                Map<String, Object> map = new HashMap<>();
                map.put("field1", random.nextInt() + "");
                map.put("field2", random.nextInt() + "");
                IndexRequest bulkItemRequest = new IndexRequest(index_name);
                bulkItemRequest.source(map);
                bulkProcessor.add(bulkItemRequest);
            }
            bulkProcessor.awaitClose(120, TimeUnit.SECONDS);
            long failure = failedBulkItemCount.get(),
                 success = bulkTotal - failure;
            System.out.println("Bulk using BulkProcessor finished with [" + success + "] requests succeeded, [" + failure + "] requests failed.");

            // Refresh the index to make written data searchable.
            RefreshRequest refreshRequest = new RefreshRequest(index_name);
            RefreshResponse refreshResponse = highClient.indices().refresh(refreshRequest, COMMON_OPTIONS);
            System.out.println("Refresh on index [" + index_name + "] successfully.");

            // Query all documents. By default, at most 10,000 results are returned.
            // To get the exact total count, call searchSourceBuilder.trackTotalHits(true).
            SearchRequest searchRequest = new SearchRequest(index_name);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            QueryBuilder queryMatchAllBuilder = new MatchAllQueryBuilder();
            searchSourceBuilder.query(queryMatchAllBuilder);
            searchRequest.source(searchSourceBuilder);
            SearchResponse searchResponse = highClient.search(searchRequest, COMMON_OPTIONS);
            long totalHit = searchResponse.getHits().getTotalHits().value;
            System.out.println("Search query match all hits [" + totalHit + "] in total.");

            // Query documents by ID.
            QueryBuilder queryByIdBuilder = new MatchQueryBuilder("_id", doc_id);
            searchSourceBuilder.query(queryByIdBuilder);
            searchRequest.source(searchSourceBuilder);
            searchResponse = highClient.search(searchRequest, COMMON_OPTIONS);
            for (SearchHit searchHit : searchResponse.getHits()) {
                System.out.println("Search query by id response [" + searchHit.getSourceAsString() + "]");
            }

            // Delete a document by ID.
            DeleteRequest deleteRequest = new DeleteRequest(index_name);
            deleteRequest.id(doc_id);
            DeleteResponse deleteResponse = highClient.delete(deleteRequest, COMMON_OPTIONS);
            System.out.println("Delete document with id [" + deleteResponse.getId() + "] successfully.");

            // Delete the index.
            DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(index_name);
            AcknowledgedResponse deleteIndexResponse = highClient.indices().delete(deleteIndexRequest, COMMON_OPTIONS);
            if (deleteIndexResponse.isAcknowledged()) {
                System.out.println("Delete index [" + index_name + "] successfully.");
            }

            highClient.close();
        } catch (Exception exception) {
            System.out.println("msg " + exception);
        }
    }
}

La salida esperada es:

Create index [lindorm_index] successfully.
Index document with id[test] successfully.
Bulk using BulkProcessor finished with [100000] requests succeeded, [0] requests failed.
Refresh on index [lindorm_index] successfully.
Search query match all hits [10000] in total.
Search query by id response [{"field1":"value1","field2":"value2"}]
Delete document with id [test] successfully.
Delete index [lindorm_index] successfully.

La consulta de coincidencia total devuelve 10,000 resultados aunque se hayan escrito 100,000 documentos. Este es el límite predeterminado de resultados. Para obtener el recuento total exacto, llame a searchSourceBuilder.trackTotalHits(true) antes de ejecutar la consulta.