How to Set Up Spring Boot 2.4 with Elasticsearch 7.8 for Bulk Indexing

Environment

Springboot 2.4.12 + Elasticsearch 7.8.0

Dependency 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>

Application Properties

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

Create Index Template

{
  "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 an index template allows each task to generate its own index without repeatedly defining mappings and settings.

Bulk Operation Base Class

public abstract class BulkOperator<T> {
    @Resource
    protected RestHighLevelClient client;

    public abstract void bulkWriter(List<T> datas, String suffix);
}

Task Bulk Service Implementation

@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("对象Task转String错误：{}", e);
            }
        });
        try {
            BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (response.status() != RestStatus.OK) {
                logger.error("批量保存失败");
            }
        } catch (IOException e) {
            logger.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";
}

Index will be created automatically.

Done!!!