elasticsearch, kibana, 6.8.18 版本下的创建索引，指定timestamp，java CRUD，maven版本等

news2024/11/14 13:53:36

ELK 这一套的版本更迭很快，
而且es常有不兼容的东西出现，
经常是搜一篇文章，看似能用，拿到我这边就不能用了。
很是烦恼。
我这边的ELK版本目前是 6.8.18，这次的操作记录一下。

（涉密内容略有删改，一看便知）

es版本信息：

{
  "name" : "es-client1",
  "cluster_name" : "xt-logs-view",
  "cluster_uuid" : "xxxxxxxxxxxx-UNg",
  "version" : {
    "number" : "6.8.18",
    "build_flavor" : "default",
    "build_type" : "docker",
    "build_hash" : "aca2329",
    "build_date" : "2021-07-28T16:06:05.232873Z",
    "build_snapshot" : false,
    "lucene_version" : "7.7.3",
    "minimum_wire_compatibility_version" : "5.6.0",
    "minimum_index_compatibility_version" : "5.0.0"
  },
  "tagline" : "You Know, for Search"
}

创建一个index，指定字段为date类型

kibana中的命令：

PUT my_index
{
  "mappings": {
    "_doc": {
      "properties": {
        "createTime": {
          "type":   "date",
          "format": "yyyy-MM-dd HH:mm:ss"
        },
        "updateTime": {
          "type":   "date",
          "format": "yyyy-MM-dd HH:mm:ss"
        }
      }
    }
  }
}

在这里插入图片描述

结合java项目中的类型

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class MyIndex implements Serializable {

	//其他字段略

    /**
     * 创建时间
     */

    @Field(type = FieldType.Date,format = DateFormat.custom,pattern = "yyyy-MM-dd HH:mm:ss")
    @JSONField(format = "yyyy-MM-dd HH:mm:ss")
    private Date createTime;

    /**
     * 更新时间
     */
    @Field(type = FieldType.Date,format = DateFormat.custom,pattern = "yyyy-MM-dd HH:mm:ss")
    @JSONField(format = "yyyy-MM-dd HH:mm:ss")
    private Date updateTime;
}

java写入es （springboot）

        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>6.8.18</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.data</groupId>
            <artifactId>spring-data-elasticsearch</artifactId>           
             <version>3.1.11.RELEASE</version>
        </dependency>

客户端配置

spring:
  data:
    elasticsearch:
      rest:
        uris: http://172.16.100.1:9200,http://172.16.100.2:9200,http://172.16.100.3:9200


@Data
@Configuration
public class ElasticSearchConfig {

    @Value("${spring.data.elasticsearch.rest.uris}")
    private String[] uris;


    @Bean
    public RestHighLevelClient restHighLevelClient() {

        //单机版
        //return new RestHighLevelClient(RestClient.builder(new HttpHost(host, port, "http")));

        //集群版
        HttpHost[] httpHosts = Arrays.stream(uris).map(HttpHost::create).toArray(HttpHost[]::new);

        //集群版
        RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(httpHosts));

        return client;
    }
}

java代码写入es

package cn.xxxxx.service.impl;


import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * @author zss
 * @date 2024年7月16日
 * @Email:
 */
@Slf4j
@Service
public class EsMyIndexServiceImpl implements EsMyIndexService {

    private static String ES_INDEX_NAME = "my_index";
    private static String ES_DOC_TYPE = "_doc";

    @Autowired
    RestHighLevelClient restHighLevelClient;


    @Override
    public void saveXxxx(MyIndex myIndex) {

        // 创建索引请求对象
        IndexRequest indexRequest = new IndexRequest(ES_INDEX_NAME, ES_DOC_TYPE);
        indexRequest.source(JSON.toJSONString(myIndex), XContentType.JSON);
        indexRequest.timeout(TimeValue.timeValueSeconds(1));
        indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
        //数据为存储而不是更新
        indexRequest.create(false);
        //indexRequest.id(mqRecord.getId() + "");
        // 执行增加文档
        restHighLevelClient.indexAsync(indexRequest, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
            @Override
            public void onResponse(IndexResponse indexResponse) {
                ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
                if (shardInfo.getFailed() > 0) {
                    for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                        log.error("mqRecord {} 存入ES时失败，原因为：{}", JSON.toJSONString(mqRecord), failure.getCause());
                    }
                }
            }

            @Override
            public void onFailure(Exception e) {
                log.error("mqRecord {} 存入es时异常，数据信息为", JSON.toJSONString(mqRecord), e);
            }
        });
    }

}