How to Bulk Insert Data into Elasticsearch with Spring Boot and Index Templates

Environment

Spring Boot 2.3.12 RELEASE with Elasticsearch 7.8.0.

The task is executed manually, and each execution should add generated data to a different index.

Configuration

<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
  </dependency>
  <dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>7.8.0</version>
  </dependency>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
  </dependency>
</dependencies>

spring.elasticsearch.index.task=t_task_
spring.elasticsearch.rest.uris=http://localhost:9200

Create Index Template

Use the following template (PUT /_template/t_task_tml) to define index patterns, settings, and mappings for fields taskId, taskName, and createTime.

{
    "index_patterns": ["t_task_*"],
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 0,
      "index": {
        "refresh_interval": "60s",
        "translog": {"sync_interval": "20s"}
      }
    },
    "mappings": {
      "_source": {"enabled": true},
      "properties": {
        "taskId": {"type": "keyword"},
        "taskName": {"type": "keyword"},
        "createTime": {"type": "date", "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"}
      }
    }
}

Creating the template allows each task to create a separate index without repeating mapping and setting information.

Bulk Operation

Define an abstract BulkOperator class and a concrete TaskBulkService that extends it.

public abstract class BulkOperator<T> {
    @Resource
    protected RestHighLevelClient client;
    public abstract void bulkWriter(List<T> datas, String suffix);
}

@Service
public class TaskBulkService extends BulkOperator<Task> {
    private static final Logger logger = LoggerFactory.getLogger(TaskBulkService.class);
    @Value("${spring.elasticsearch.index.task:t_task_}")
    private String index_name;

    @Override
    public void bulkWriter(List<Task> datas, String suffix) {
        BulkRequest bulkRequest = new BulkRequest();
        bulkRequest.timeout(TimeValue.timeValueMinutes(1));
        ObjectMapper objectMapper = new ObjectMapper();
        datas.forEach(task -> {
            try {
                bulkRequest.add(new IndexRequest(index_name + suffix)
                    .id(task.getId())
                    .source(objectMapper.writeValueAsString(task), XContentType.JSON));
            } catch (JsonProcessingException e) {
                logger.error("Object Task to String error: {}", e);
            }
        });
        try {
            BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (response.status() != RestStatus.OK) {
                logger.error("Bulk save failed");
            }
        } catch (IOException e) {
            logger.error("Bulk save error: {}", e);
        }
    }
}

Test Endpoint

@GetMapping("/es/bulk")
public Object bulk(String suffix) {
    List<Task> datas = new ArrayList<>();
    for (int i = 0; i < 10; i++) {
        Task task = new Task();
        task.setTaskId("taskId " + i);
        task.setTaskName("taskName " + i);
        task.setCreateTime("createTime " + i);
        datas.add(task);
    }
    bulkOperator.bulkWriter(datas, suffix);
    return "success";
}

The index will be created automatically.

Done!