一、前言

最近工作中有这样一个ElasticSearch（以下简称ES）写入的场景，Flink解决完数据实时写入ES。当初须要将一批历史数据通过Flink加载到到ES，有两个点须要保障：

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

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

二、代码实现及验证

代码实现

申请写数据时退出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. 将工夫版本比拟的字段值固定为第一次执行程序的工夫戳，测验数据是否更新胜利。

验证后果如下图：


三、总结

由截图可看到，第一步和第二步都能执行胜利，第三步执行会呈现版本抵触的异样，依据提醒很不便能辨认出起因，即ElasticSearch进阶篇(一)--版本控制中得出的论断，应用version和version_type=EXTERNAL进行版本控制时，只有要写入文档的版本号大于已有文档的版本号能力更新胜利。

案例代码参考：elasticsearch_demo