Java中ES5.3批量插入_bulk实现方案

...

<dependency>

    <groupId>org.elasticsearch.client</groupId>

    <artifactId>transport</artifactId>

    <version>5.5.3</version>

</dependency>

<dependency>

    <groupId>org.elasticsearch.plugin</groupId>

    <artifactId>transport-netty3-client</artifactId>

    <version>5.5.3</version>

</dependency>

<dependency>

    <groupId>org.elasticsearch.client</groupId>

    <artifactId>x-pack-transport</artifactId>

    <version>5.5.3</version>

</dependency>

...

<repository>

    <id>elasticsearch-releases</id>

    <url>https://artifacts.elastic.co/maven</url>

    <releases>

        <enabled>true</enabled>

    </releases>

    <snapshots>

        <enabled>false</enabled>

    </snapshots>

</repository>

...

@Configuration

@Log4j2

public class ElasticSearchConfig {

    @Value("${icec.elasticsearch.host}")

    private String host;

    @Value("${icec.elasticsearch.tcpPort}")

    private int tcpPort;

    @Value("${icec.elasticsearch.clusterName}")

    private String clusterName;

    @Value("${icec.elasticsearch.username}")

    private String username;

    @Value("${icec.elasticsearch.password}")

    private String password;

    @Bean

    public TransportClient transportClient() {

        TransportClient transportClient = null;

        TransportClient preBuiltTransportClient = new PreBuiltXPackTransportClient(Settings.builder()

                .put("cluster.name", clusterName)

                .put("xpack.security.user", username + ":" + password)

                .put("client.transport.sniff", false)

                .build());

        try {

            transportClient = preBuiltTransportClient

                    .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), tcpPort));

        } catch (UnknownHostException e) {

            log.warn(e);

        }

        return transportClient;

    }

}

public void batchInsert(List<Map> datas) {

    if (CollectionUtils.isEmpty(datas)) {

        return;

    }

    BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();

    datas.forEach(data -> bulkRequestBuilder.add(transportClient.prepareIndex(ES_INDEX, ES_TYPE, (String) data.get(OrderEsConstant.ORDER_ID)).setSource(data)));

    BulkResponse bulkResponse = bulkRequestBuilder.execute().actionGet();

    List<String> failuerMessages = Arrays.asList(bulkResponse.getItems())

            .stream().filter(p -> StringUtils.isNotBlank(p.getFailureMessage()))

            .map(BulkItemResponse::getFailureMessage).collect(Collectors.toList());

    if (CollectionUtils.isNotEmpty(failuerMessages)) {

        log.info("同步失败订单->{}", failuerMessages);

    }

    log.info("批处理完成 总条数：{}", datas.size());

}

public void batchInsert(List<Map> orders) {

    StringBuilder bulkRequestBody = new StringBuilder();

    int count = 1;

    for (Map order : orders) {

        String actionMetaData = String.format("{ \"index\" : {\"_id\" : \"%s\"} }%n", order.get(OrderEsConstant.ORDER_ID));

        String orderJson = JSON.toJSONString(order, SerializerFeature.WriteNullStringAsEmpty);

        bulkRequestBody.append(actionMetaData);

        bulkRequestBody.append(orderJson);

        bulkRequestBody.append("\n");

        if (count % 5000 == 0 || count == orders.size()) {

            Response response;

            Map result = Maps.newHashMap();

            String router = String.format(BASIC_FORMAT, ES_INDEX, ES_TYPE, "_bulk");

            try {

                response = restClient.performRequest(POST, router, Collections.emptyMap(), new StringEntity(bulkRequestBody.toString(), ContentType.APPLICATION_JSON));

                result = objectMapper.readValue(response.getEntity().getContent(), Map.class);

            } catch (IOException e) {

                log.info("ES批量插入异常");

            }

            List<Map> itemResults = ((List<Map>) result.get("items")).stream().map(p -> (Map) p.get("index")).collect(Collectors.toList());

            List errorMsgs = itemResults.stream().filter(p -> 200 != (Integer) p.get("status")).map(this::getErrorMsg).collect(Collectors.toList());

            if (CollectionUtils.isNotEmpty(errorMsgs)) {

                log.warn("数据插入失败 -> {}", errorMsgs);

            }

            log.info("批处理完成，总条数： {}", itemResults.size());

        }

        count++;

    }

}