Elasticsearch基于聚合创建新索引：如何仅保留特定条件数据

你的_reindex请求中添加的聚合节点仅会执行统计分析，不会对要复制的文档进行筛选，所以才会把原索引的全量数据都复制到新索引里。要实现过滤需求，需要根据你的具体条件选择以下方法：

方法1：直接添加查询条件过滤数据 #

如果你的“特定条件”是简单的字段匹配（比如字段值相等、范围筛选等），直接在source节点下添加query模块即可：

POST /_reindex?wait_for_completion=false
{
  "source": {
    "index": "as.event-154",
    "query": {
      // 示例：只保留ancestor_id为"target_id"的文档
      "term": {
        "ancestor_id": "target_id"
      }
      // 也可以组合多条件，比如同时筛选时间范围
      /*
      "bool": {
        "must": [
          {"term": {"ancestor_id": "target_id"}},
          {"range": {"event_time": {"gte": "2024-01-01T00:00:00"}}}
        ]
      }
      */
    }
  },
  "dest": {
    "index": "as.event-154new1"
  }
}

方法2：基于聚合结果过滤（保留特定分组的文档） #

如果需求是只保留满足聚合条件的分组下的文档（比如只保留ancestor_id分组数量大于指定值的文档），需要分两步操作：

步骤1：执行聚合查询，提取符合条件的ancestor_id列表 #

POST /as.event-154/_search?size=0
{
  "aggs": {
    "group": {
      "terms": {
        "field": "ancestor_id",
        "size": 10000 // 调整size确保覆盖所有目标分组
      },
      "aggs": {
        "filter_valid_groups": {
          "bucket_selector": {
            "buckets_path": {
              "group_count": "_count"
            },
            "script": "params.group_count > 5" // 这里写分组的过滤条件，比如分组数量大于5
          }
        }
      }
    }
  }
}

从返回结果的aggregations.group.buckets中提取所有key值（即符合条件的ancestor_id）。

步骤2：在_reindex中用terms查询过滤这些ID #

将步骤1得到的ID列表填入query模块：

POST /_reindex?wait_for_completion=false
{
  "source": {
    "index": "as.event-154",
    "query": {
      "terms": {
        "ancestor_id": ["id_1", "id_2", "id_3"] // 替换成步骤1提取的ID列表
      }
    }
  },
  "dest": {
    "index": "as.event-154new1"
  }
}

补充：按分组保留TopN文档 #

如果你的需求是每个ancestor_id只保留最新的10条文档，需要先通过聚合获取目标文档ID，再执行_reindex：

1. 先获取每个分组的Top10文档ID：

POST /as.event-154/_search?size=0
{
  "aggs": {
    "group": {
      "terms": {
        "field": "ancestor_id",
        "size": 1
      },
      "aggs": {
        "top_docs": {
          "top_hits": {
            "size": 10,
            "_source": ["_id"], // 仅提取文档ID
            "sort": [{"event_id.keyword": {"order": "desc"}}]
          }
        }
      }
    }
  }
}

2. 提取所有top_docs.hits.hits._id，然后用ids查询过滤：

POST /_reindex?wait_for_completion=false
{
  "source": {
    "index": "as.event-154",
    "query": {
      "ids": {
        "values": ["doc_id_1", "doc_id_2", ...] // 替换成提取的文档ID列表
      }
    }
  },
  "dest": {
    "index": "as.event-154new1"
  }
}

内容的提问来源于stack exchange，提问作者bala n