Spring Data Elasticsearch Массовое индексирование/удаление – миллионы записей ⇐ JAVA

[Anonymous]

Я использую Spring Data Elasticsearch 4.2.5, у нас есть задание, которое выполняет ETL (извлечение, преобразование и загрузку данных) в определенную таблицу базы данных. Я индексирую эти данные с помощью Elasticsearch во время выполнения задания. Данные будут в миллионах записей и более. В настоящее время я делаю индекс на каждой итерации. Я читал, что использование индекса elasticsearch на каждой итерации может занять некоторое время. Я хотел использовать что-то вроде массового индекса, но для этого мне нужно добавить объект indexQuery в список. Добавление миллионов записей в список и выполнение массового индексирования может вызвать проблемы с памятью.
Мне нужно применить аналогичный процесс для удаления. Когда записи удаляются на основе какого-то общего идентификатора, мне нужно удалить связанные эластичные документы, и это также будет в миллионах и более.
Можно ли как-нибудь очень быстро индексировать/удалять для этого требование? Любая помощь будет очень признательна, и поправьте меня, если мое понимание неверно.
ИНДЕКСИРОВАНИЕ
for (Map.Entry key : objectDetailsHashMap.entrySet()) {
indexDocument(elasticsearchOperations, key, oPath);
// other code to insert data in db table...
}

private void indexDocument(ElasticsearchOperations elasticsearchOperations,
Map.Entry key, String oPath) {
String docId = "" + key.getValue().getCatalogId() + key.getValue().getObjectId();

byte[] nameBytes = key.getValue().getName();
byte[] physicalNameBytes = key.getValue().getPhysicalName();
byte[] definitionBytes = key.getValue().getDefinition();
byte[] commentBytes = key.getValue().getComment();

IndexQuery indexQuery = new IndexQueryBuilder()
.withId(docId)
.withObject(new MetadataSearch(
key.getValue().getObjectId(),
key.getValue().getCatalogId(),
key.getValue().getParentId(),
key.getValue().getTypeCode(),
key.getValue().getStartVersion(),
key.getValue().getEndVersion(),
nameBytes != null ? new String(nameBytes, StandardCharsets.UTF_8) : "-",
physicalNameBytes != null ? new String(physicalNameBytes, StandardCharsets.UTF_8) : "-",
definitionBytes != null ? new String(definitionBytes, StandardCharsets.UTF_8) : "-",
commentBytes != null ? new String(commentBytes, StandardCharsets.UTF_8) : "-",
oPath
))
.build();

elasticsearchOperations.index(indexQuery, IndexCoordinates.of("portal_idx"));
}

УДАЛЕНИЕ
private void deleteElasticDocuments(String catalogId) {
String queryText = martServerContext.getQueryCacheInstance().getQuery(QUERY_PORTAL_GET_OBJECTS_IN_PORTAL_BY_MODEL);
MapSqlParameterSource mapSqlParameterSource = new MapSqlParameterSource();
mapSqlParameterSource.addValue("cId", Integer.parseInt(catalogId));
namedParameterJdbcTemplate.query(queryText, mapSqlParameterSource, (resultSet -> {
int objectId = resultSet.getInt(O_ID);
String docId = catalogId + objectId;
elasticsearchOperations.delete(docId, IndexCoordinates.of("portal_idx"));
}));
}


Подробнее здесь: https://stackoverflow.com/questions/693 ... of-records