Node.js操作Elasticsearch设置唯一字段及删除重复数据方法

Elasticsearch原生不提供关系型数据库那样的字段级唯一约束配置，无法直接通过mapping参数实现自动去重。这一设计是因为ES是分布式架构，字段级全局唯一校验需要跨节点做分布式事务检查，性能损耗极高，因此官方始终推荐用_id天然唯一的特性实现去重，原子性有保障，无并发问题，性能最优。

你当前的代码没有手动指定文档_id，ES每次写入都会自动生成随机ID，因此同title的文档每次写入都会生成新条目，最终出现重复。

最优实现方案：用title生成固定文档ID #

ES的_id是索引级全局唯一的，相同_id写入时默认会更新原文档，不会生成重复条目。只需要把title做标准化处理（去前后空格、哈希避免特殊字符）后作为文档ID写入，即可实现同title文档不会重复插入的效果。
如果需要“仅在文档不存在时写入，不覆盖已有文档”，只需要增加op_type: 'create'参数，同ID文档已存在时ES会直接返回版本冲突，不会修改原有数据。
另外你原有代码对async函数额外包装一层Promise属于不必要的反模式，直接返回async执行结果即可。修改后的写入代码如下：

const crypto = require('crypto');
// 基于title生成固定唯一文档ID，做trim避免前后空格导致判定差异，md5避免特殊字符不符合ID规范
const getDocIdByTitle = (title) => {
  return crypto.createHash('md5').update(title.trim()).digest('hex');
}

async function insertManual(manual) {
  const docId = getDocIdByTitle(manual.title);
  const result = await client.index({
    index: 'completeindexthree',
    id: docId,
    op_type: 'create', // 不需要覆盖原有数据就加这行，需要更新最新数据就删掉这行
    body: {
      brand: manual.brand,
      category: manual.category,
      url: manual.url,
      title: manual.title.trim(),
      parsingData: new Date().toString()
    }
  })
  await client.indices.refresh({index: 'completeindexthree'})
  return result;
}

注意：不要用“先查询title是否存在再决定是否插入”的逻辑，该逻辑在多实例/多并发爬虫场景下存在竞态条件，依然会出现重复写入，上面的_id+op_type:create方案是ES原子操作，无并发问题。

实现逻辑为：先通过聚合查询找出所有存在重复的title分组，每个分组按规则保留1条有效文档，其余文档通过批量接口删除。操作前务必备份索引，避免误删数据。

注意前提 #

聚合和精确匹配需要用到title的精确值字段：如果你的title字段是text类型（默认分词类型），需要确保mapping中存在title.keyword的keyword类型子字段；如果title本身就是keyword类型，代码中把title.keyword替换为title即可。

实现代码 #

async function cleanDuplicateByTitle() {
  const indexName = 'completeindexthree';
  // 1. 聚合查询所有重复title分组
  const aggRes = await client.search({
    index: indexName,
    size: 0,
    body: {
      aggs: {
        duplicate_groups: {
          terms: {
            field: 'title.keyword',
            min_doc_count: 2, // 只筛选文档数>=2的重复分组
            size: 100000 // 根据你索引的总title量级调整，要大于等于索引内唯一title总数
          },
          aggs: {
            group_docs: {
              top_hits: {
                size: 100, // 单个title下最多返回的重复文档数，根据实际重复峰值调整
                // 排序规则：按parsingData倒序，即最新爬取的排在最前，会被保留；要保留最早的就改为asc
                sort: [{ parsingData: { order: 'desc' } }],
                _source: false
              }
            }
          }
        }
      }
    }
  });

  // 2. 整理待删除的文档ID列表
  const toDeleteIds = [];
  const buckets = aggRes.aggregations.duplicate_groups.buckets;
  for (const bucket of buckets) {
    const docIds = bucket.group_docs.hits.hits.map(hit => hit._id);
    // 每个分组保留排序后的第一条，其余加入删除列表
    toDeleteIds.push(...docIds.slice(1));
  }

  if (toDeleteIds.length === 0) {
    console.log('未检测到按title判定的重复文档');
    return;
  }

  // 3. 分批批量删除，ES单次bulk操作建议不超过1000条，避免请求过大超时
  const batchSize = 1000;
  for (let i = 0; i < toDeleteIds.length; i += batchSize) {
    const batchIds = toDeleteIds.slice(i, i + batchSize);
    const bulkBody = [];
    for (const id of batchIds) {
      bulkBody.push({ delete: { _index: indexName, _id: id } });
    }
    const bulkRes = await client.bulk({
      body: bulkBody,
      refresh: true
    });
    if (bulkRes.errors) {
      console.error(`第${Math.floor(i/batchSize) + 1}批删除出现错误`, bulkRes.items);
    } else {
      console.log(`已完成第${Math.floor(i/batchSize) + 1}批重复数据删除，共处理${batchIds.length}条`);
    }
  }
  console.log(`重复数据清理完成，累计删除重复文档${toDeleteIds.length}条`);
}

// 执行清理，确认逻辑无误后再放开注释运行
// cleanDuplicateByTitle().catch(err => console.error('清理任务执行失败', err));

清理完成后，必须把写入逻辑替换为问题1中的固定ID写入方案，否则下次爬虫运行依然会生成新的重复数据。

内容的提问来源于stack exchange，提问作者nikita cohen