java批量写入es数据库(java es批量写入)

简述：

        为了测试一个转接平台的数据库适合mysql还是es，或是mongodb，先进行mysql和es的对比，当前想通过java批量将mysql中的数据导入到es中。此篇文章作为基础，写一个批量插入es的demo。

基础环境版本：

pom.xml

    
        
            aliyun
            https://maven.aliyun.com/repository/central
        
    

    
        
            co.elastic.clients
            elasticsearch-java
            8.17.1
        
    

demo主代码

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.bulk.IndexOperation;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class ElasticsearchBulkExample {
    public static void main(String[] args) throws IOException {
        // 创建 Elasticsearch 客户端
        RestClient restClient = RestClient.builder(
                new HttpHost("localhost", 9200, "http")
        ).build();

        RestClientTransport transport = new RestClientTransport(
                restClient, new JacksonJsonpMapper()
        );

        ElasticsearchClient esClient = new ElasticsearchClient(transport);

        // 准备批量数据
        List operations = new ArrayList();
        for (int i = 1; i <= 10; i++) {
            final String docId = String.valueOf(i);
            Map document = Map.of(
                    "id", i,
                    "name", "User " + i,
                    "age", 20 + i
            );
            operations.add(BulkOperation.of(op -> op
                    .index(IndexOperation.of(idx -> idx
                            .index("jsl")
                            .id(docId)
                            .document(document)
                    ))));
        }

        // 批量写入请求
        BulkRequest bulkRequest = new BulkRequest.Builder()
                .operations(operations)
                .build();

        // 执行批量操作
        BulkResponse bulkResponse = esClient.bulk(bulkRequest);

        if (!bulkResponse.errors()) {
            System.out.println("批量写入成功!");
        } else {
            System.err.println("批量写入过程中存在错误!");
        }

        // 关闭资源
        transport.close();
        restClient.close();
    }
}

运行结果