??工作中遇到這樣一個需求場景：由于ES數(shù)據(jù)庫中歷史數(shù)據(jù)過多，占用太多的磁盤空間，需要定期地進行清理，在一定程度上可以釋放磁盤空間，減輕磁盤空間壓力。
??在經(jīng)過調(diào)研之后發(fā)現(xiàn)，某服務(wù)項目每周產(chǎn)生的數(shù)據(jù)量已經(jīng)達到千萬級別，單日將近能產(chǎn)生兩百萬的數(shù)據(jù)量寫入到 ES 數(shù)據(jù)庫中，平均每個小時最少產(chǎn)生 10w+ 條數(shù)據(jù)，加上之前的歷史數(shù)據(jù)，目前生產(chǎn)環(huán)境 ES 數(shù)據(jù)量已經(jīng)達到兩億一千四百八十萬的數(shù)據(jù)。并且隨著當(dāng)前業(yè)務(wù)量的爆發(fā)式增長，數(shù)據(jù)增長量急劇飆升，在未來一年內(nèi)每周產(chǎn)生的數(shù)據(jù)量有望達到 3kw-5kw 左右。
??因此，對 ES 數(shù)據(jù)庫中歷史數(shù)據(jù)進行清理勢在必行，為了能夠釋放磁盤空間，并且還要保證業(yè)務(wù)方能夠進行日常問題的排查定位，決定從兩個月前的數(shù)據(jù)開始清理，方案如下：
- 編寫定時任務(wù)，每天凌晨三點清理兩個月前的那一天數(shù)據(jù)，之所以選擇凌晨三點是因為在 Grafana 查看了生產(chǎn)環(huán)境的集群監(jiān)控情況，凌晨兩點至四點之間的集群、索引的查詢以及寫入 QPS 都比較低。
- 清理一天的數(shù)據(jù)時，根據(jù)時間段進行清理，每個小時清理一次，避免內(nèi)存中存放太多的數(shù)據(jù)，導(dǎo)致內(nèi)存溢出。
- 清理 ES 數(shù)據(jù)時，需要先查詢出數(shù)據(jù)，而 ES 默認(rèn)最多只能查詢 1w 條數(shù)據(jù)，如果當(dāng)次需要刪除的數(shù)據(jù)量超過 1w 條，普通的查詢操作無法完全刪除數(shù)據(jù)。因此，需要采用滾動查詢的方式，滾動查詢結(jié)果保持時間需要設(shè)置合理，不能太長，否則也可能會導(dǎo)致內(nèi)存溢出。
根據(jù)以上的思路方案，設(shè)計的定時清理ES歷史數(shù)據(jù)代碼如下：
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.search.ClearScrollRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.Date;
/**
* 清理ES歷史數(shù)據(jù)定時任務(wù)
*/
@Component
public class CleanESHistoryDataTask {
private static final Logger LOGGER = LoggerFactory.getLogger(CleanESHistoryDataTask.class);
@Resource
private RestHighLevelClient restHighLevelClient;
/**
* 根據(jù)索引名稱刪除當(dāng)前日期兩個月前的那一天的歷史文檔數(shù)據(jù)
* @param jobContext
*/
@Scheduled
public void cleanESHistoryData(JobContext jobContext) {
// jobContext為定時任務(wù)中回傳數(shù)據(jù)
String indexName = jobContext.getData();
if (StringUtils.isBlank(indexName)) {
LOGGER.warn("ES索引名稱不能為空!");
return;
}
long startTimeMillis = System.currentTimeMillis();
String twoMonthsAgoDate = DateTool.format(DateUtils.addMonths(new Date(), -1), DateTool.DF_DAY);
try {
String startTimeStr = twoMonthsAgoDate + " 00:00:00";
// 初始化時間，形如2023-08-06 00:00:00
Date initialStartTime = DateTool.parse(startTimeStr, DF_FULL);
// 每次循環(huán)清理一個小時歷史文檔數(shù)據(jù),循環(huán)24次清理完一天的歷史文檔數(shù)據(jù)
for (int i = 0; i < 24; i++) {
Date startTime = initialStartTime;
startTime = DateUtils.addHours(startTime, i);
Date endTime = DateUtils.addHours(startTime, 1);
LOGGER.info("正在清理索引:[{}],時間:{} 至 {}的歷史文檔數(shù)據(jù)...", indexName, DateTool.format(startTime, DF_FULL), DateTool.format(endTime, DF_FULL));
long currentStartTimeMillis = System.currentTimeMillis();
// 指定操作的索引庫
SearchRequest searchRequest = new SearchRequest(indexName);
// 構(gòu)造查詢條件，指定查詢的時間范圍，每次最多寫入1000條數(shù)據(jù)至內(nèi)存，減輕服務(wù)器內(nèi)存壓力
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().query(QueryBuilders.rangeQuery("createTimeStr.keyword")
.from(DateTool.format(startTime, DF_FULL))
.to(DateTool.format(endTime, DF_FULL)))
.size(1000);
// 設(shè)置滾動查詢結(jié)果在內(nèi)存中的過期時間為1min
Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L));
// 將滾動以及構(gòu)造的查詢條件放入查詢請求
searchRequest.scroll(scroll).source(searchSourceBuilder);
SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
// 記錄要滾動的ID
String scrollId = searchResponse.getScrollId();
SearchHit[] hits = searchResponse.getHits().getHits();
while (hits != null && hits.length > 0) {
// 創(chuàng)建批量處理請求對象
BulkRequest bulkRequest = new BulkRequest();
for (SearchHit hit : hits) {
DeleteRequest deleteRequest = new DeleteRequest(indexName, hit.getId());
bulkRequest.add(deleteRequest);
}
// 執(zhí)行批量刪除請求操作
restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
// 構(gòu)造滾動查詢條件，繼續(xù)滾動查詢
SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
scrollRequest.scroll(scroll);
searchResponse = restHighLevelClient.scroll(scrollRequest, RequestOptions.DEFAULT);
scrollId = searchResponse.getScrollId();
hits = searchResponse.getHits().getHits();
}
// 當(dāng)前滾動查詢結(jié)束,清除滾動,釋放服務(wù)器內(nèi)存資源
ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
clearScrollRequest.addScrollId(scrollId);
restHighLevelClient.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
LOGGER.info("清理索引:[{}],時間:{} 至 {}的歷史文檔數(shù)據(jù)成功,耗時{}ms", indexName, DateTool.format(startTime, DF_FULL), DateTool.format(endTime, DF_FULL), (System.currentTimeMillis() - currentStartTimeMillis));
}
LOGGER.info("[cleanESHistoryData] 定時任務(wù)-清理索引:[{}],時間:{}的歷史文檔數(shù)據(jù)成功,耗時{}ms", indexName, twoMonthsAgoDate, (System.currentTimeMillis() - startTimeMillis));
} catch (Exception e) {
LOGGER.error(String.format("[cleanESHistoryData] 定時任務(wù)-清理索引:[%],時間:%的歷史文檔數(shù)據(jù)失敗,耗時%ms", indexName, twoMonthsAgoDate, (System.currentTimeMillis() - startTimeMillis)), e);
}
}
}
其中，需要注意以下幾點：
- 在 Java 中對 ES 進行操作，這里使用的是 ES 的高級客戶端組件
RestHighLevelClient。 -
@Scheduled注解為自研定時任務(wù)工具注解，外界無法使用，在使用定時任務(wù)時需要自己選擇合適的定時任務(wù)框架。 -
DateTool工具類為自研工具類，外界同樣無法使用，在以上代碼段中就是用于對java.util.Date類型進行轉(zhuǎn)換為字符串，DF_FULL和DateTool.DF_DAY均是常量，它們的值分別為yyyy-MM-dd HH:mm:ss和yyyy-MM-dd。
文章來源：http://www.zghlxwxcb.cn/news/detail-632172.html
??通過觀察監(jiān)控可以發(fā)現(xiàn)，在凌晨三點執(zhí)行定時任務(wù)清理 ES 歷史數(shù)據(jù)期間，集群、索引查詢 QPS 以及 CPU 利用率指標(biāo)都明顯飆升。因此，清理 ES 數(shù)據(jù)時一定要避開流量高峰期，避免在流量高峰期清理數(shù)據(jù)時造成資源實例宕機，造成生產(chǎn)事故。文章來源地址http://www.zghlxwxcb.cn/news/detail-632172.html
到了這里，關(guān)于【已解決】Java 中使用 ES 高級客戶端庫 RestHighLevelClient 清理百萬級規(guī)模歷史數(shù)據(jù)的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
