54 · ingest pipeline 攝取管道寫(xiě)入時(shí)加工數(shù)據(jù)階段第六階段 / 進(jìn)階專題ESingest pipeline processors | PostgreSQLBEFORE INSERT觸發(fā)器 / ETL 轉(zhuǎn)換1. 概念ingest pipeline讓文檔在寫(xiě)入索引之前先過(guò)一串processor處理器做加工補(bǔ)字段、改類型、拆分、脫敏、算派生值、調(diào)用 enrich 補(bǔ)維表……全部在 ES 側(cè)完成應(yīng)用端不用寫(xiě)這些轉(zhuǎn)換邏輯。一句話寫(xiě)入時(shí)的輕量 ETL跑在 ES 協(xié)調(diào)節(jié)點(diǎn)上。常用 processorset、rename、convert、date、grok正則解析日志、split、gsub、scriptPainless、enrich補(bǔ)維表、remove。2. PostgreSQL 對(duì)照-- BEFORE INSERT 觸發(fā)器寫(xiě)入前加工CREATEFUNCTIONfill_defaults()RETURNStriggerAS$$BEGINNEW.created_at :now();NEW.amount :NEW.price*NEW.qty;RETURNNEW;END;$$LANGUAGEplpgsql;CREATETRIGGERt BEFOREINSERTONsalesFOR EACH ROWEXECUTEFUNCTIONfill_defaults();ingest pipeline 就是 ES 版的「寫(xiě)入前觸發(fā)器 / ETL 轉(zhuǎn)換」但配置化、可復(fù)用。3. ES DSL3.1 定義 pipelinePUT _ingest/pipeline/sales_pipeline { description: 寫(xiě)入前補(bǔ)字段、算金額、打時(shí)間戳, processors: [ { set: { field: ingested_at, value: {{_ingest.timestamp}} } }, { convert: { field: qty, type: integer } }, { script: { source: ctx.amount ctx.price * ctx.qty } }, { remove: { field: tmp_raw, ignore_missing: true } } ], on_failure: [ // 出錯(cuò)兜底別丟數(shù)據(jù) { set: { field: _ingest_error, value: {{ _ingest.on_failure_message }} } } ] }3.2 用 pipeline 寫(xiě)入# 單條寫(xiě)入時(shí)指定 POST sales_idx/_doc?pipelinesales_pipeline { price: 10, qty: 5, tmp_raw: x } # 寫(xiě)完得到 amount50、ingested_at... # 或把它設(shè)為索引默認(rèn)之后所有寫(xiě)入自動(dòng)走 PUT sales_idx/_settings { index.default_pipeline: sales_pipeline }3.3 用 _simulate 先試跑強(qiáng)烈推薦POST _ingest/pipeline/sales_pipeline/_simulate { docs: [ { _source: { price: 10, qty: 5 } } ] }4. Spring Boot 實(shí)現(xiàn)ComponentpublicclassDoc54IngestPipeline{AutowiredprivateElasticsearchClientelasticsearchClient;/** 創(chuàng)建/更新 pipeline定義外置成 JSONwithJson 直灌 */publicvoidputPipeline(Stringid,StringpipelineJson)throwsIOException{elasticsearchClient.ingest().putPipeline(p-p.id(id).withJson(newStringReader(pipelineJson)));}/** 寫(xiě)入時(shí)指定 pipeline也可在 bulk 的每個(gè)操作上帶 pipeline */publicvoidindexWithPipeline(Stringindex,Stringpipeline,MapString,Objectdoc)throwsIOException{elasticsearchClient.index(i-i.index(index).pipeline(pipeline)// ← 走攝取管道加工.document(doc));}}ingest 客戶端在elasticsearchClient.ingest()。bulk也支持在請(qǐng)求或每條操作上指定pipeline。5. 坑與最佳實(shí)踐先_simulate再上線processor 順序、字段名錯(cuò)很常見(jiàn)模擬能省大量返工。一定配on_failure默認(rèn)某條處理失敗會(huì)整條寫(xiě)入失敗兜底把錯(cuò)誤記下來(lái)別丟數(shù)據(jù)。default_pipeline方便但隱蔽設(shè)成索引默認(rèn)后所有寫(xiě)入都會(huì)走排查問(wèn)題記得想到它。grok解析日志很強(qiáng)但慢正則別寫(xiě)太貪婪結(jié)構(gòu)化數(shù)據(jù)優(yōu)先用dissect。ingest 跑在協(xié)調(diào)/ingest 節(jié)點(diǎn)重加工會(huì)占 CPU大流量考慮專用 ingest 節(jié)點(diǎn)。和 enrich 配合寫(xiě)入時(shí)補(bǔ)維表字段第 51 篇就是通過(guò) ingest 的enrichprocessor。