分类： Elasticsearch

起因

一次需求中涉及两个索引，其中一个是本次新建的索引，需要将原始ES索引中的部分数据导入到当前新建的索引中。

分析

对于新所以的数据初始化有两个思路，因为我们的数据都是在HIVE中清洗完成之后再打到ElasticSearch的，所以可以从HIVE中捞取对应的数据重新打到ElasticSearch即可。但是这里有一个局限性的问题，因为HIVE表数量巨大，我们只保留的近7天的清洗结果，如果按照需求的时间跨度则需要重新清洗近3个月的数据，之后再将数据打到ES。

但是这样的话又会有另外一个问题，HIVE表数据在清洗过程中使用到的数据是会经常变动的，比如某个商品最近几天是挂靠在A品类下，但是上个月因为做活动是挂靠在B品类下的，这就导致事后隔了一段时间之后重新清洗的结果无法反应出当时这条数据正确的关联关系的情况。如果产品产品认可当前方案，那么也不失为可行方案之一。

那么，既然我们在原始的索引中有这部分数据，所以自然使用reindex命令来完成这一操作就最为方便了

代码

POST _reindex?wait_for_completion=false&slices=5&refresh
{
  "source": {
    "index": "SOURCE_ES_INDEX",
    "type": "SOURCE_ES_TYPE",
    "query": {
      "range": {
        "last_modify_time": {
          "gte": "2023-01-01 00:00:00",
          "lte": "2023-07-24 23:59:59"
        }
      }
    }
  },
  "dest": {
    "index": "TARGET_ES_INDEX",
    "type": "TARGET_ES_TYPE"
  }
}

简单说下这条语句，公司生产环境使用的ES版本为5.4.2的，所以除了index参数外，还需要有对应的type参数，但是Elasticsearch在 7.X版本中去除了type的概念，所以如果你的ES版本较新的话，可以不用这个type参数，即：

在ES的日常使用中，需要使用到Nested结构存储数个同级的子节点数据，例如一条主订单下的N条子订单的数据。

新增更新操作

现在，假设我们在ES中有这样一条数据

PUT /celebrities/_doc/114
{
    "user" : "Kun",
    "post_date" : "2009-11-15T14:12:12",
    "message" : "Idol who has been practicing for two and a half years",
    "skills":[
        {
            "name":"sing",
            "skill_level":"A"
        },
        {
            "name":"jump",
            "skill_level":"S"
        },
        {
            "name":"rap",
            "skill_level":"SS"
        }
    ]
}

我们需要往skills的Nested数组中添加一个新的节点，节点的name为“consecutive five whips”，则可以这么写

POST /celebrities/_doc/114
{
    "script": {
        "source": "if (ctx._source.skills == null) {List ls = new ArrayList();ls.add(params.skill);ctx._source.skills = ls;} else {ctx._source.skills.add(params.skill);}",
        "lang": "painless",
        "params": {
            "skill": {
                "name": "consecutive five whips",
                "skill_level": "SSS"
            }
        }
    }
}

得到返回结果，表明执行成功

{
    "_index": "celebrities",
    "_id": "114",
    "_version": 6,
    "result": "updated",
    "_shards": {
        "total": 2,
        "successful": 1,
        "failed": 0
    },
    "_seq_no": 6,
    "_primary_term": 1
}

通过代码可以简单的看出逻辑，如果_source.skills为null，则创建一个新的ArrayList，并将参数中的skill节点的内容放进去，最后赋值给_source.skills，反之如果不为null的话，则直接往里加入当前的skill字段的内容。

if (ctx._source.skills == null) {
    List ls = new ArrayList();
    ls.add(params.skill);
    ctx._source.skills = ls;
} else {
    ctx._source.skills.add(params.skill);
}

要先判断是否为null，否则无法调用add方法，并会抛出一个异常

"caused_by": {
                "type": "null_pointer_exception",
                "reason": "cannot access method/field [add] from a null def reference"
}

起因

生产环境一个最近新上了一个清理ES历史数据的定时任务，执行_delete_by_query语句。在检查系统异常日志的时候，发现这块功能报了数个异常，elastic: Error 409 (Conflict)。于是带着这个这个问题展开探索一番

代码中使用的方法就是最基本DSL根据查询条件进行删除的语句

POST /ES_INDEX/_delete_by_query
{
  "query": {
    "term": {
      "data_date": "${dataDate}"
    }
  }
}

在执行_delete_by_query期间，为了删除匹配到的所有文档，多个搜索请求是按顺序执行的。每次找到一批文档时，将会执行相应的批处理请求来删除找到的全部文档。如果搜索或者批处理请求被拒绝，_delete_by_query根据默认策略对被拒绝的请求进行重试（最多10次）。达到最大重试次数后，会造成_delete_by_query请求中止，并且会在failures字段中响应 所有的故障。已经删除的仍会执行。换句话说，该过程没有回滚，只有中断。

原因

Elasticsearch 会在删除文档前使用查询时获取的 internal 版本号和当前文档版本号进行对比。如果在查询和删除操作之间文档发生更改，则会导致版本冲突并且删除操作失败，报 elastic: Error 409 (Conflict) 错误。