54 · ingest pipeline 摄取管道写入时加工数据阶段第六阶段 / 进阶专题ESingest pipeline processors | PostgreSQLBEFORE INSERT触发器 / ETL 转换1. 概念ingest pipeline让文档在写入索引之前先过一串processor处理器做加工补字段、改类型、拆分、脱敏、算派生值、调用 enrich 补维表……全部在 ES 侧完成应用端不用写这些转换逻辑。一句话写入时的轻量 ETL跑在 ES 协调节点上。常用 processorset、rename、convert、date、grok正则解析日志、split、gsub、scriptPainless、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 DSL3.1 定义 pipelinePUT _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?pipelinesales_pipeline { price: 10, qty: 5, tmp_raw: x } # 写完得到 amount50、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定义外置成 JSONwithJson 直灌 */publicvoidputPipeline(Stringid,StringpipelineJson)throwsIOException{elasticsearchClient.ingest().putPipeline(p-p.id(id).withJson(newStringReader(pipelineJson)));}/** 写入时指定 pipeline也可在 bulk 的每个操作上带 pipeline */publicvoidindexWithPipeline(Stringindex,Stringpipeline,MapString,Objectdoc)throwsIOException{elasticsearchClient.index(i-i.index(index).pipeline(pipeline)// ← 走摄取管道加工.document(doc));}}ingest 客户端在elasticsearchClient.ingest()。bulk也支持在请求或每条操作上指定pipeline。5. 坑与最佳实践先_simulate再上线processor 顺序、字段名错很常见模拟能省大量返工。一定配on_failure默认某条处理失败会整条写入失败兜底把错误记下来别丢数据。default_pipeline方便但隐蔽设成索引默认后所有写入都会走排查问题记得想到它。grok解析日志很强但慢正则别写太贪婪结构化数据优先用dissect。ingest 跑在协调/ingest 节点重加工会占 CPU大流量考虑专用 ingest 节点。和 enrich 配合写入时补维表字段第 51 篇就是通过 ingest 的enrichprocessor。
54 · ingest pipeline 摄取管道写入时加工数据阶段第六阶段 / 进阶专题ESingest pipeline processors | PostgreSQLBEFORE INSERT触发器 / ETL 转换1. 概念ingest pipeline让文档在写入索引之前先过一串processor处理器做加工补字段、改类型、拆分、脱敏、算派生值、调用 enrich 补维表……全部在 ES 侧完成应用端不用写这些转换逻辑。一句话写入时的轻量 ETL跑在 ES 协调节点上。常用 processorset、rename、convert、date、grok正则解析日志、split、gsub、scriptPainless、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 DSL3.1 定义 pipelinePUT _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?pipelinesales_pipeline { price: 10, qty: 5, tmp_raw: x } # 写完得到 amount50、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定义外置成 JSONwithJson 直灌 */publicvoidputPipeline(Stringid,StringpipelineJson)throwsIOException{elasticsearchClient.ingest().putPipeline(p-p.id(id).withJson(newStringReader(pipelineJson)));}/** 写入时指定 pipeline也可在 bulk 的每个操作上带 pipeline */publicvoidindexWithPipeline(Stringindex,Stringpipeline,MapString,Objectdoc)throwsIOException{elasticsearchClient.index(i-i.index(index).pipeline(pipeline)// ← 走摄取管道加工.document(doc));}}ingest 客户端在elasticsearchClient.ingest()。bulk也支持在请求或每条操作上指定pipeline。5. 坑与最佳实践先_simulate再上线processor 顺序、字段名错很常见模拟能省大量返工。一定配on_failure默认某条处理失败会整条写入失败兜底把错误记下来别丢数据。default_pipeline方便但隐蔽设成索引默认后所有写入都会走排查问题记得想到它。grok解析日志很强但慢正则别写太贪婪结构化数据优先用dissect。ingest 跑在协调/ingest 节点重加工会占 CPU大流量考虑专用 ingest 节点。和 enrich 配合写入时补维表字段第 51 篇就是通过 ingest 的enrichprocessor。
