我正在嘗試在 Python 中創(chuàng)建一個 API 以從 ES 中提取數(shù)據(jù)并將其提供給數(shù)據(jù)倉庫。數(shù)據(jù)是實時的并且每秒都會被填充，所以我將創(chuàng)建一個近乎實時的管道。當(dāng)前的 URL 格式是{{url}}/{{index}}/_search，我發(fā)送的測試負(fù)載是：{ "from" : 0, "size" : 5}在下一次刷新時，它將使用有效負(fù)載進(jìn)行拉取：{ "from" : 6, "size" : 5}以此類推，直到達(dá)到記錄總數(shù)。PROD 環(huán)境有大約 250M 行，我將大小設(shè)置為每次提取 10K。我很擔(dān)心，因為我不知道這些記錄是否在 ES 中被重新排序。目前，有一個使用用戶生成的時間戳的插件，但它存在缺陷，因為有時由于 json 可用于在 ES 中提取的延遲以及時間的生成方式可能導(dǎo)致文檔被跳過。有誰知道使用提取數(shù)據(jù)時的默認(rèn)排序是什么/_search？
2 回答
有只小跳蛙
TA貢獻(xiàn)1824條經(jīng)驗 獲得超8個贊
在與我的同事考慮后，我們決定實施和使用_ingestAPI，而不是在 ES 中創(chuàng)建一個管道，該管道在每個文檔上插入服務(wù)器文檔攝取日期。
腳步：
創(chuàng)建時間戳管道
PUT _ingest/pipeline/timestamp_pipeline
{
? "description" : "Inserts timestamp field for all documents",
? "processors" : [
? ? {
? ? ? "set" : {
? ? ? ? "field": "insert_date",
? ? ? ? "value": "{{_ingest.timestamp}}"
? ? ? }
? ? }
? ]
}
更新索引以添加新的默認(rèn)字段
PUT /*/_settings
{
? "index" : {
? ? "default_pipeline": "timestamp_pipeline"
? }
}
在 Python 中，我會_scroll像這樣使用 API：
? ? es = Elasticsearch(cfg.esUrl, port = cfg.esPort, timeout = 200)
? ? doc = {
? ? ? "query": {
? ? ? ? "range": {
? ? ? ? ? "insert_date": {
? ? ? ? ? ? "gte": lastRowDateOffset
? ? ? ? ? }
? ? ? ? }
? ? ? }
? ? }
? ? res = es.search(
? ? ? ? index = Index,
? ? ? ? sort = "insert_date:asc",
? ? ? ? scroll = "2m",
? ? ? ? size = NumberOfResultsPerPage,
? ? ? ? body = doc
? ? )
lastRowDateOffset最后一次跑步的日期在哪里
ITMISS
TA貢獻(xiàn)1871條經(jīng)驗 獲得超8個贊
我想您正在尋找的是一個流式傳輸/更改 API，@Val 在這里對此進(jìn)行了很好的描述，還有一個開放的功能請求。
與此同時，您不能真正依賴size和from參數(shù)——您可能會進(jìn)行冗余查詢并在重復(fù)項到達(dá)您的數(shù)據(jù)倉庫之前對其進(jìn)行處理。
另一種選擇是在這方面跳過 ES 并直接流式傳輸?shù)絺}庫嗎？我的意思是，在給定時間之前拍攝一次 ES 快照（這樣您就可以保留歷史數(shù)據(jù)），將其提供給倉庫，然后直接從您獲取數(shù)據(jù)的地方流式傳輸?shù)絺}庫。
附錄
AFAIK 默認(rèn)排序是按插入日期。但是沒有內(nèi)部_insertTime或類似的東西。你可以使用游標(biāo)——它被稱為滾動，這是一個 py實現(xiàn)。但這是從“最新”文檔到“第一個”文檔，反之亦然。所以它會給你所有現(xiàn)有的文檔，但我不太確定你滾動時新添加的文檔。然后你想再次運行滾動，這是次優(yōu)的。
您還可以預(yù)先排序您的索引，當(dāng)結(jié)合滾動時，它應(yīng)該非常適合您的用例。
添加回答
舉報
0/150
提交
取消