Elasticsearch使用脚本upsert文档时索引不存在报错如何解决

报错核心原因有两个：

• 当目标按月滚动的新索引不存在、对应文档首次写入时，ctx._source对象本身为空，原有脚本直接访问ctx._source['status']会触发空指针异常，打断Elasticsearch默认的索引自动创建、文档初始化流程。
• 原有脚本的优先级逻辑存在缺陷：如果CLOSED状态消息先写入成功，后续OPEN状态消息到达时会走全量覆盖分支，直接擦除CLOSED状态写入的closing_ts、field1等字段，不符合预设的CLOSED状态优先级规则。

1. 调整Logstash Elasticsearch output配置 #

开启脚本upsert支持，告知Elasticsearch即使文档/索引不存在，也要执行脚本完成初始化，配置片段如下：

output {
  elasticsearch {
    hosts => ["你的ES集群地址"]
    # 按月滚动索引配置
    index => "你的业务索引名-%{+YYYY.MM}"
    # 用payload中的id作为文档唯一_id
    document_id => "%{id}"
    action => "update"
    # 核心配置：开启脚本upsert能力，文档不存在时不跳过脚本执行
    scripted_upsert => true
    script_type => "inline"
    script => "
      // 首次写入、文档不存在场景：直接用当前事件初始化文档
      if (ctx._source == null) {
        ctx._source = params.event;
      } else {
        boolean incomingIsClosed = params.event.status == 'CLOSED';
        boolean existingIsClosed = ctx._source.status == 'CLOSED';

        if (incomingIsClosed) {
          // 当前写入的是CLOSED状态：高优先级，覆盖所有同名字段，独有字段保留
          for (key in params.event.keySet()) {
            ctx._source[key] = params.event[key];
          }
        } else if (!existingIsClosed) {
          // 已有文档不是CLOSED、当前写入的是OPEN状态：正常全量覆盖
          for (key in params.event.keySet()) {
            ctx._source[key] = params.event[key];
          }
        } else {
          // 已有文档是CLOSED、当前写入的是OPEN状态：仅补全不存在的字段，不覆盖已有内容
          for (key in params.event.keySet()) {
            if (ctx._source[key] == null) {
              ctx._source[key] = params.event[key];
            }
          }
        }
      }
    "
  }
}

2. 边界场景验证 #

• 新索引首条写入：不管首条消息是OPEN还是CLOSED状态，脚本检测到ctx._source为空会直接初始化文档，ES自动创建对应月份的索引，无报错。
• OPEN先到、CLOSED后到：OPEN正常初始化文档，CLOSED到达时按高优先级规则覆盖status、field1等字段，保留已写入的opening_ts，新增closing_ts，结果符合预期。
• CLOSED先到、OPEN后到：CLOSED正常初始化文档，OPEN到达时检测到已有状态为CLOSED，仅补全缺失的opening_ts字段，不会覆盖已有的CLOSED相关字段，结果符合预期。

内容的提问来源于stack exchange，提问作者K-You