Elasticsearch агрегации для дубликатов

Question

Я надеюсь, что кто-нибудь может помочь мне разобраться с этой проблемой.Занимался этим несколько дней и чувствую, что ударил стену.Буду очень признателен за любую помощь по этой теме.Вот проблема -

Я заполняю записи адресов в индексе.Проблема в том, что источник отправляет тонну дубликатов, которые нуждаются в дедупликации на основе адреса и города (к сожалению, кроме этих текстовых полей нет PK, поэтому я понимаю, что это не будет самый быстрый запрос, но я согласен саспект производительности на данный момент) Эта дедупликация не должна происходить в режиме реального времени, и мы делаем это раз в день.Есть около 16 миллионов документов, и около 800 тысяч из них выглядят как дубликаты.Вот что я сделал до сих пор -

1. Прежде всего, я игнорирую все строки, в которых адрес или город нулевые.Я также рассматриваю только те документы, в которых для параметра record_status установлено значение «активный».Подробнее об этом флаге см. В шаге 3.
2. Найдите дубликаты, используя следующий запрос агрегации
3. Как только я получу результаты обратно в виде фрагментов, я отправлю запрос на обновление обратно в ES, чтобы перевернуть атрибутс именем record_status = 'inactive', поэтому я исключаю их в следующем вызове агрегации.
4. По какой-то причине он не захватывает все дубликаты документов при создании сегментов.Я вижу, что конкретный комбо адрес + город существует в индексе 31 раз, но в списке только 6 из них.Что еще хуже, когда я включаю этот конкретный адрес вручную, он показывает все 31 документов.Что дает ??

Вот запрос DSL, который я использую

{
  "size": 0,
  "aggs": {
"addresses": {
  "terms": {
    "script" : {
        "lang": "painless",
        "source": "if ((doc['display_address.keyword'].value != null && doc['city.keyword'].value != null) && doc['record_status.keyword'].value == 'active') {return doc['display_address.keyword'].value + doc['city.keyword'].value;} else {return null;}"
  },
  "size": 500,
  "exclude": "null"
},
"aggs": {
  "my_filter": {
    "bucket_selector": {
      "buckets_path": {
        "the_doc_count": "_count"
      },
      "script": "params.the_doc_count > 1"
    }
  },
  "final_docs": {
    "top_hits": {
      "sort": [
        {
          "list_date": {
            "order": "desc"
          }
        }
      ],
      "_source": {
        "includes": [
          "_id",
          "list_date",
          "display_address",
          "city"
        ]
      },
      "size": 100
    }
  }
}
  }
}
}

Этот запрос выше дает мне неверные данные для нескольких адресов.Например, для 123 Main St, Детройт возвращается только с 6 документами.

Вместо этого, если я просто добавлю

`"include": "123 Main st"` 

в мою агрегацию, вдруг я вижу 31 дубликат для 123Main st, Детройт