基于Spring Boot REST API优化大日志文件Elasticsearch写入性能

我正在构建一个Spring Boot REST API，接收MultipartFile类型的日志文件作为参数，将文件内容存入独立的Elasticsearch索引，每行日志对应一个文档。实现步骤为：

• 创建索引；
• 逐行读取文件；
• 将每行内容存入索引。
  使用的Elasticsearch版本为8.6。

遇到的问题：处理大文件（最大可达200MB）时速度极慢，即使是远小于100MB的普通文件也耗时数小时。

以下是我使用BufferedReader和Elasticsearch Java API Client编写的代码：

// 创建索引
String indexName = "newIndex";

CreateIndexResponse createResponse = client.indices().create(
        new CreateIndexRequest.Builder()
                .index(indexName)
                .build()
);
// 逐行读取并保存
try (BufferedReader br = new BufferedReader(new InputStreamReader(file.getInputStream()))) {
    String line = "";
    while ((line = br.readLine()) != null) {
        if (!line.trim().isEmpty()) {
            LogFile log = new LogFile();
            log.setMessage(line);
            IndexResponse resp = client.index(i -> i
                    .index(indexName)
                    .document(log));
        }
    }
}

// LogFile类代码
@Document(indexName = "log-model")
@Data
public class LogFile {

    @Id
    private String id;
    
    @Field(type = FieldType.Text)
    private String message;

}

需求：如何优化现有代码？是否有类似Logstash的工具可缩短大文件加载时间？

一、代码层面优化 #

核心问题是逐条发送Elasticsearch请求，产生大量网络往返开销，这是性能瓶颈的主要原因。可从以下几点优化：

1. 使用批量请求（Bulk API）
   Elasticsearch的Bulk API允许一次性提交多个文档操作，大幅减少网络请求次数。建议积累一定数量的文档（比如1000条）后再批量提交：

String indexName = "newIndex";
client.indices().create(new CreateIndexRequest.Builder().index(indexName).build());

try (BufferedReader br = new BufferedReader(new InputStreamReader(file.getInputStream()))) {
    String line;
    List<IndexOperation> operations = new ArrayList<>();
    int batchSize = 1000; // 根据服务器性能调整，范围500-2000

    while ((line = br.readLine()) != null) {
        if (!line.trim().isEmpty()) {
            LogFile log = new LogFile();
            log.setMessage(line);
            // 添加到批量操作列表
            operations.add(new IndexOperation.Builder().index(indexName).document(log).build());
            
            // 达到批次大小就提交
            if (operations.size() >= batchSize) {
                client.bulk(b -> b.index(indexName).operations(operations));
                operations.clear();
            }
        }
    }
    // 提交剩余的文档
    if (!operations.isEmpty()) {
        client.bulk(b -> b.index(indexName).operations(operations));
    }
}

2. 调整索引刷新频率
   默认Elasticsearch每秒刷新索引，会影响写入性能。批量写入前可临时关闭自动刷新，完成后再恢复：

// 批量写入前关闭自动刷新
client.indices().putSettings(p -> p
        .index(indexName)
        .settings(s -> s.put("refresh_interval", "-1")));

// 执行批量写入操作...

// 写入完成后恢复默认刷新频率
client.indices().putSettings(p -> p
        .index(indexName)
        .settings(s -> s.put("refresh_interval", "1s")));

3. 临时关闭副本
   如果是一次性导入数据，可临时将索引副本数设为0，避免副本同步开销，导入完成后再恢复：

// 导入前关闭副本
client.indices().putSettings(p -> p
        .index(indexName)
        .settings(s -> s.put("number_of_replicas", 0)));

// 写入完成后恢复副本数（比如1）
client.indices().putSettings(p -> p
        .index(indexName)
        .settings(s -> s.put("number_of_replicas", 1)));

4. 优化文件读取
   使用Files.lines结合并行流加速读取，注意批量操作时的线程安全：

try (Stream<String> lines = Files.lines(Paths.get(file.getOriginalFilename()))) {
    // 配合批量逻辑处理...
}

二、工具层面替代方案 #

如果不想自行优化代码，这类场景可直接使用现成工具：

1. Logstash
   Logstash自带file输入插件和elasticsearch输出插件，配置简单，内置批量处理、多线程等优化，适合大文件导入。典型配置示例：

input {
  file {
    path => "/path/to/your/log/file.log"
    start_position => "beginning" # 从头开始读取文件
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "newIndex"
    bulk_size => 1000 # 批量提交大小
  }
}

2. Filebeat + Logstash
   如果是持续的日志导入需求，可使用Filebeat采集日志文件，发送给Logstash批量处理后写入Elasticsearch，性能和稳定性更优。

3. 自定义批量脚本
   用Python/Shell脚本读取日志文件，生成Elasticsearch Bulk格式的数据，直接调用Bulk接口导入，适合快速批量操作。

内容的提问来源于stack exchange，提问作者BEY MEHREZ