Elasticsearch通过_update_by_query为历史数据添加GeoIP字段报错如何解决？

1. _update_by_query 接口本身不支持直接传入 processors 作为顶层参数，你的写法不符合ES API规范，导致解析异常。
2. 你请求体中的script结构是运行时字段的定义格式，不是_update_by_query支持的脚本格式，同时你使用的doc变量是查询上下文的语法，文档更新上下文需要用ctx变量访问字段。

推荐你把GeoIP处理和后续时间字段计算逻辑封装为独立的Ingest Pipeline，再通过_update_by_query指定调用该管道实现存量数据补全，步骤如下：

步骤1：创建处理管道 #

执行以下请求创建包含所有需求逻辑的摄入管道：

PUT _ingest/pipeline/add_geo_and_time_fields
{
  "processors": [
    {
      "geoip": {
        "field": "client_ip",
        "target_field": "geo",
        "database_file": "GeoLite2-City.mmdb",
        "properties": ["continent_name", "country_iso_code", "country_name", "city_name", "timezone", "location"]
      }
    },
    {
      "script": {
        "source": """
          ZonedDateTime ts = ctx['@timestamp'].withZoneSameInstant(ZoneId.of(ctx.geo.timezone));
          ctx.day_of_week = ts.getDayOfWeek().getValue();
          ctx.hour_of_day = ts.getHour();
          ctx.office_hours = (ctx.day_of_week >= 1 && ctx.day_of_week <=5 && ctx.hour_of_day >=8 && ctx.hour_of_day <=18) ? 1 : -1;
        """
      }
    }
  ]
}

步骤2：调用_update_by_query执行存量更新 #

执行以下请求触发全量存量数据更新：

POST my_index*/_update_by_query?pipeline=add_geo_and_time_fields
{
  "refresh": true
}

如果你的数据量较大，可添加参数开启异步执行和并行处理，避免接口超时：

POST my_index*/_update_by_query?pipeline=add_geo_and_time_fields&wait_for_completion=false&slices=auto
{
  "refresh": true
}

该请求会返回任务ID，你可以通过任务API查看执行进度。

• 执行全量更新前建议对目标索引创建快照备份，避免误操作导致数据异常
• 正式执行前可先选取小批量测试数据验证管道逻辑是否符合预期，确认无误后再执行全量更新

内容的提问来源于stack exchange，提问作者Saba Far