Elasticsearch Java批量索引优化方案
在进行文档写入操作时，频繁的单条请求会显著影响性能。当需要处理海量数据时，采用逐条插入的方式显然不具可行性。
传统批量操作示例：
POST /_bulk
{ "delete": { "_index": "website", "_type": "blog", "_id": "123" }}
{ "create": { "_index": "website", "_type": "blog", "_id": "123" }}
{ "title": "My first blog post" }
{ "index": { "_index": "website", "_type": "blog" }}
{ "title": "My second blog post" }
{ "update": { "_index": "website", "_type": "blog", "_id": "123", "_retry_on_conflict" : 3} }
{ "doc" : {"title" : "My updated blog post"} }
标准批量处理实现：
@Test
public void batchInsert() throws IOException {
BulkRequestHandler bulkHandler = client.prepareBulk();
bulkHandler.add(client.prepareIndex("social", "post", "1001")
.setSource(jsonBuilder()
.startObject()
.field("author", "李四")
.field("timestamp", new Date())
.field("content", "关于网络事件的深度分析")
.endObject()
)
);
bulkHandler.add(client.prepareIndex("social", "post", "1002")
.setSource(jsonBuilder()
.startObject()
.field("author", "王五")
.field("timestamp", new Date())
.field("content", "体育赛事的实时报道")
.endObject()
)
);
BulkResponse response = bulkHandler.get();
if (response.hasFailures()) {
// 处理失败项
}
}
该方案存在局限性，如批量阈值、数据量限制、执行间隔等参数均需手动配置。以下为增强型处理方案：
优化批量处理实现：
@Test
public void optimizedBatch() throws Exception {
BulkProcessor processor = BulkProcessor.builder(client, new BulkProcessor.Listener() {
public void beforeBulk(long id, BulkRequest request) {
System.out.println("准备提交" + request.numberOfActions() + "条记录");
}
public void afterBulk(long id, BulkRequest request, BulkResponse response) {
System.out.println("成功处理" + request.numberOfActions() + "条记录");
}
public void afterBulk(long id, BulkRequest request, Throwable error) {
System.out.println("处理" + request.numberOfActions() + "条记录失败");
}
})
.setBulkActions(5000)
.setBulkSize(new ByteSizeValue(500, ByteSizeUnit.MB))
.setFlushInterval(TimeValue.timeValueSeconds(10))
.setConcurrentRequests(2)
.setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(50), 5))
.build();
Map data = new HashMap<>();
data.put("text", "异步批量测试数据");
processor.add(new IndexRequest("news", "article", "2001").source(data));
processor.add(new IndexRequest("news", "article", "2002").source(data));
processor.flush();
processor.awaitClose(5, TimeUnit.MINUTES);
}
