ElasticSearch版本控制--java实现

1、前言

最近工做中有这样一个ElasticSearch（如下简称ES）写入的场景，Flink处理完数据实时写入ES。如今须要将一批历史数据经过Flink加载到到ES，有两个点须要保证：

1. 对于历史数据，ES已有文档，则舍弃旧数据，ES没有则插入历史数据。
2. 对于新数据，能对现有的ES数据进行更新。

参考ElasticSearch进阶篇(一)--版本控制,能够使用ES的版本实现该需求的开发。

2、代码实现及验证

代码实现

请求写数据时加入version和version_type参数，主要代码以下：

IndexRequest indexRequest = Requests.indexRequest()
                .index(indexName)
                .id("1")
                // 指定版本比较的业务字段，具体业务具体分析，通常取时间戳较为合适
                .version(Long.parseLong(dataMap1.get("create").toString()))
                // 指定使用外部版本号
                .versionType(VersionType.EXTERNAL)
                .source(dataMap);

验证

验证demo可以使用当前时间的时间戳做为版本比较依据。验证思路以下：

1. 运行demo程序，在当前时间戳下，插入一条数据，经过kibana等工具检验数据是否插入成功。并记录当前的时间戳。
2. 更改某些字段值对数据进行更新，再次运行程序，检验数据是否更新成功。
3. 将时间版本比较的字段值固定为第一次执行程序的时间戳，检验数据是否更新成功。

验证结果以下图：

3、总结

由截图可看到，第一步和第二步都能执行成功，第三步执行会出现版本冲突的异常，根据提示很方便能识别出缘由，即ElasticSearch进阶篇(一)--版本控制中得出的结论，使用version和version_type=EXTERNAL进行版本控制时，只有要写入文档的版本号大于已有文档的版本号才能更新成功。

案例代码参考：elasticsearch_demo