Spring Boot中如何将批量JSON文件索引到Elasticsearch

前置准备 #

• 本地待导入的JSON源文件：文档结构需与ES目标索引的字段映射提前做兼容校验，优先采用每行单文档的NDJSON格式；如果是根节点为数组的大JSON文件，需提前通过流式解析拆分结构，避免全量加载导致内存溢出
• Spring Boot项目已引入ES客户端依赖：可选择Spring Data Elasticsearch Starter，或直接引入Elasticsearch官方RestHighLevelClient依赖
• ES集群基础参数就绪：包括集群访问地址、端口、认证信息、提前创建完成的目标索引（预定义字段映射，避免自动映射导致的字段类型错乱、搜索异常）

具体实现步骤 #

1. 配置ES客户端连接 #

• 在项目配置文件中写入ES连接参数，适配大文件批量导入的超时场景

elasticsearch:
  host: ES集群访问地址
  port: 9200
  username: 集群访问账号
  password: 集群访问密码

• 初始化可用于批量写入的ES客户端Bean，调整超时参数适配长耗时批量请求：

@Configuration
public class EsClientConfig {
    @Value("${elasticsearch.host}")
    private String esHost;
    @Value("${elasticsearch.port}")
    private int esPort;
    @Value("${elasticsearch.username}")
    private String esUser;
    @Value("${elasticsearch.password}")
    private String esPwd;

    @Bean
    public RestHighLevelClient restHighLevelClient() {
        ClientConfiguration config = ClientConfiguration.builder()
                .connectedTo(esHost + ":" + esPort)
                .withBasicAuth(esUser, esPwd)
                .withConnectTimeout(Duration.ofSeconds(10))
                .withSocketTimeout(Duration.ofMinutes(1))
                .build();
        return new RestHighLevelClient(config);
    }
}

2. 本地JSON文件流式读取 #

• 采用逐行流式读取方式处理JSON文件，不要一次性将整个文件加载到内存中
• 逐行校验JSON结构合法性，跳过格式错误的行并记录异常内容，避免单条脏数据中断全量导入流程
• 核心读取逻辑参考：

private static final int BATCH_SIZE = 1000;
private final List<String> pendingDocs = new ArrayList<>();

public void processJsonFile(String filePath) throws IOException {
    ObjectMapper objectMapper = new ObjectMapper();
    try (BufferedReader reader = new BufferedReader(new FileReader(filePath))) {
        String line;
        while ((line = reader.readLine()) != null) {
            // 校验JSON结构合法性
            if (objectMapper.readTree(line) != null) {
                pendingDocs.add(line);
                // 攒够单批次数据就触发导入
                if (pendingDocs.size() == BATCH_SIZE) {
                    batchWriteToEs(pendingDocs);
                    pendingDocs.clear();
                }
            }
        }
        // 处理最后剩余的不足一个批次的数据
        if (!pendingDocs.isEmpty()) {
            batchWriteToEs(pendingDocs);
        }
    }
}

3. ES批量索引导入实现 #

• 单批次写入的文档量控制在500-2000区间，平衡导入速率与ES集群负载，避免过大批次导致集群写入拒绝
• 为批量请求增加简易重试逻辑，覆盖网络波动、集群临时不可用的异常场景
• 核心导入逻辑参考：

@Autowired
private RestHighLevelClient esClient;
private static final String TARGET_INDEX = "目标索引名称";

private void batchWriteToEs(List<String> docBatch) throws IOException {
    BulkRequest bulkRequest = new BulkRequest();
    for (String docJson : docBatch) {
        // 可自定义文档ID，不设置则由ES自动生成随机唯一ID
        IndexRequest indexReq = new IndexRequest(TARGET_INDEX)
                .source(docJson, XContentType.JSON);
        bulkRequest.add(indexReq);
    }
    // 执行批量写入
    BulkResponse bulkResp = esClient.bulk(bulkRequest, RequestOptions.DEFAULT);
    // 识别批次内的写入失败记录，可做单独的补写处理
    if (bulkResp.hasFailures()) {
        log.error("批量导入异常：{}", bulkResp.buildFailureMessage());
    }
}

实操关键注意事项 #

• 正式全量导入前，先取10-20条样本文档做小批量导入验证，确认字段映射匹配、文档内容可正常检索后再启动全量流程
• 大文件导入过程中实时监控ES集群的CPU使用率、写入线程池队列长度、磁盘IO指标，避免全量写入对集群承载的其他线上业务造成影响
• 导入完成后做双向校验：统计本地JSON文件的有效文档行数，与ES索引通过_count接口返回的总文档数做对比，确认无漏写、错写
• 针对特别大的JSON文件，可通过分片拆分文件的方式并行导入，进一步缩短全量导入的耗时

内容的提问来源于stack exchange，提问作者Jeevan Suvarna