54 · ingest pipeline 摄取管道（写入时加工数据）

阶段：第六阶段 / 进阶专题
ES：ingest pipeline + processors | PostgreSQL：BEFORE INSERT触发器 / ETL 转换

1. 概念

ingest pipeline让文档在写入索引之前先过一串processor（处理器）做加工：
补字段、改类型、拆分、脱敏、算派生值、调用 enrich 补维表……全部在 ES 侧完成，
应用端不用写这些转换逻辑。

一句话：写入时的轻量 ETL，跑在 ES 协调节点上。

常用 processor：set、rename、convert、date、grok（正则解析日志）、
split、gsub、script（Painless）、enrich（补维表）、remove。

2. PostgreSQL 对照

-- BEFORE INSERT 触发器：写入前加工CREATEFUNCTIONfill_defaults()RETURNStriggerAS$$BEGINNEW.created_at :=now();NEW.amount :=NEW.price*NEW.qty;RETURNNEW;END;$$LANGUAGEplpgsql;CREATETRIGGERt BEFOREINSERTONsalesFOR EACH ROWEXECUTEFUNCTIONfill_defaults();

ingest pipeline 就是 ES 版的「写入前触发器 / ETL 转换」，但配置化、可复用。

3. ES DSL

3.1 定义 pipeline

PUT _ingest/pipeline/sales_pipeline { "description": "写入前补字段、算金额、打时间戳", "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": [ // 出错兜底：别丢数据 { "set": { "field": "_ingest_error", "value": "{{ _ingest.on_failure_message }}" } } ] }

3.2 用 pipeline 写入

# 单条写入时指定 POST sales_idx/_doc?pipeline=sales_pipeline { "price": 10, "qty": "5", "tmp_raw": "x" } # 写完得到 amount=50、ingested_at=... # 或把它设为索引默认，之后所有写入自动走 PUT sales_idx/_settings { "index.default_pipeline": "sales_pipeline" }

3.3 用 _simulate 先试跑（强烈推荐）

POST _ingest/pipeline/sales_pipeline/_simulate { "docs": [ { "_source": { "price": 10, "qty": "5" } } ] }

4. Spring Boot 实现

@ComponentpublicclassDoc54IngestPipeline{@AutowiredprivateElasticsearchClientelasticsearchClient;/** 创建/更新 pipeline：定义外置成 JSON，withJson 直灌 */publicvoidputPipeline(Stringid,StringpipelineJson)throwsIOException{elasticsearchClient.ingest().putPipeline(p->p.id(id).withJson(newStringReader(pipelineJson)));}/** 写入时指定 pipeline（也可在 bulk 的每个操作上带 pipeline） */publicvoidindexWithPipeline(Stringindex,Stringpipeline,Map<String,Object>doc)throwsIOException{elasticsearchClient.index(i->i.index(index).pipeline(pipeline)// ← 走摄取管道加工.document(doc));}}

ingest 客户端在elasticsearchClient.ingest()。bulk也支持在请求或每条操作上指定pipeline。

5. 坑与最佳实践

1. 先_simulate再上线：processor 顺序、字段名错很常见，模拟能省大量返工。
2. 一定配on_failure：默认某条处理失败会整条写入失败；兜底把错误记下来别丢数据。
3. default_pipeline方便但隐蔽：设成索引默认后，所有写入都会走，排查问题记得想到它。
4. grok解析日志很强但慢：正则别写太贪婪；结构化数据优先用dissect。
5. ingest 跑在协调/ingest 节点：重加工会占 CPU，大流量考虑专用 ingest 节点。
6. 和 enrich 配合：写入时补维表字段（第 51 篇）就是通过 ingest 的enrichprocessor。