「Painless 脚本与 Ingest 管道：从数据清洗到管道编排」

Painless 是 Elasticsearch 内置的脚本语言，专为在文档索引前完成数据清洗、字段转换与条件改写而生；Ingest 管道则把多个处理器编排成一条流水线。理解脚本上下文与管道处理器的分工，是构建稳定数据接入链路的前提。本文从脚本基础讲起，逐步覆盖管道编排、调试与性能安全。

1. Painless 脚本基础

一句话总结： Painless 是 ES 内置的轻量脚本语言，语法贴近 Java，用于在索引、查询与聚合的特定上下文里做字段级计算。

1.1 Painless 的定位

Painless 于 5.0 版本取代 Groovy 成为 Elasticsearch 唯一的脚本语言。它编译成 JVM 字节码，运行在受限沙箱中，禁止文件 IO、网络访问与反射，保证恶意脚本无法逃逸。与 Groovy 相比，Painless 启动更快、更安全、更可控。

// Painless 基本表达式
int a = 10;
int b = 20;
int sum = a + b;
return sum;

1.2 第一个可运行脚本

在 _script API 中可以直接求值脚本，这是验证语法最快的途径：

curl -X POST "localhost:9200/_scripts/painless/_execute" -H "Content-Type: application/json" -d'
{
  "script": {
    "source": "return params.a * params.b + 1;",
    "params": { "a": 3, "b": 4 }
  }
}
'

脚本通过 params 对象传入外部参数，避免把可变值硬编码进源码——这样脚本可被缓存复用，性能更好，也防止拼接注入。

1.3 类型与语法要点

Painless 支持 Java 的大部分类型与运算符：整型、浮点型、字符串、数组、List、Map，以及三目运算、lambda、def 动态类型。def 在编译期不检查类型，灵活但损耗性能，生产脚本应显式声明类型。

def name = "Alice";
Map user = [ "age": 30, "city": "Beijing" ];
List tags = [ "vip", "new_user" ];
return name + ":" + user.city + ", tags=" + tags.size();

2. 脚本上下文与 API

一句话总结： 脚本必须运行在特定上下文里，每个上下文提供不同的文档对象与可访问字段，Ingest 上下文用 ctx 操作文档。

2.1 上下文分类

脚本分为上下文：ingest（管道处理器中的脚本）、search（查询中的 script 查询/排序）、filter（脚本过滤）、update（文档更新）、painless_test（执行 API）。不同上下文暴露的对象不同，例如 ingest 上下文提供 ctx，search 上下文提供 doc。

2.2 Ingest 上下文与 ctx

在 Ingest 管道中，脚本处理器通过 ctx 对象读写当前文档字段。ctx 是一个 Map，直接修改即可改变即将落库的文档：

// 字段级转换：把金额从分转成元，并打上来源标签
if (ctx.containsKey("amount_cents")) {
  ctx.amount_yuan = ctx.amount_cents / 100.0;
}
ctx.source = "payment-gateway";
ctx.tags = ["cleaned", "verified"];

2.3 查询与聚合上下文

查询上下文通过 doc 读取 Doc Values，速度极快但不能读 _source；需要全字段的排序计算时用 _source 但更慢。聚合脚本同样用 doc 读取字段，配合 params 传参实现可复用的排序键。

// 查询脚本：按自定义公式排序
"script": {
  "source": "double click = doc['click_count'].value; double view = doc['view_count'].value; return click / (view + 1);",
  "lang": "painless"
}

3. 字段转换与条件处理

一句话总结： 脚本处理器可对字段做读写、判空、条件分支与集合运算，是管道中最灵活的加工单元。

3.1 字段读写与判空

ctx 是可变 Map，新增字段直接赋值，读取不存在的字段返回 null。写入前务必判空，否则 NullPointer 会中断整条管道：

String email = ctx.email;
if (email == null || email.isEmpty()) {
  ctx.email_valid = false;
} else {
  ctx.email_domain = email.substring(email.indexOf('@') + 1);
  ctx.email_valid = true;
}

3.2 条件分支与三元运算

管道内除了脚本处理器，每个处理器还可以挂 if 条件。if 返回布尔表达式，true 才执行该处理器，用于按业务规则跳过处理：

{
  "processors": [
    { "set": { "field": "risk", "value": "high", "if": "ctx.amount > 10000" } },
    { "set": { "field": "risk", "value": "normal", "if": "ctx.amount <= 10000" } }
  ]
}

3.3 集合运算与循环

列表字段的去重、排序、聚合经常在脚本里完成，例如给标签列表去重并统一小写：

List cleaned = new ArrayList();
for (String t : ctx.tags) {
  String lower = t.toLowerCase(Locale.ROOT);
  if (!cleaned.contains(lower)) {
    cleaned.add(lower);
  }
}
ctx.tags = cleaned;

4. Ingest 管道处理器

一句话总结： 管道由一组处理器按顺序执行，每个处理器完成一类原子操作，set/rename/remove/convert 是最常用组合。

4.1 常用处理器清单

Ingest 提供大量内置处理器：set 设值、rename 重命名、remove 删除、convert 类型转换、lowercase/uppercase、trim、gsub 正则替换、split 切分、join 拼接、date 日期解析、grok/dissect 文本解析。能用内置处理器解决的，就不要写脚本。

4.2 定义一条完整管道

PUT _ingest/pipeline/normalize-log
{
  "description": "日志归一化管道",
  "processors": [
    { "lowercase": { "field": "level" } },
    { "convert": { "field": "response_time", "type": "float", "ignore_missing": true } },
    { "rename": { "field": "msg", "target_field": "message", "ignore_missing": true } },
    { "remove": { "field": ["raw_line", "_meta"], "ignore_missing": true } }
  ]
}

4.3 管道执行顺序与错误处理

处理器按数组顺序串行执行，前一个失败默认终止整条管道并抛异常。通过 on_failure 可以把失败文档导向降级处理，例如记录到死信索引：

PUT _ingest/pipeline/log-with-fallback
{
  "processors": [
    { "grok": { "field": "message", "patterns": ["%{TIMESTAMP_ISO8601:ts} %{LOGLEVEL:level} %{GREEDYDATA:rest}"] } }
  ],
  "on_failure": [
    { "set": { "field": "parse_failed", "value": true } },
    { "set": { "field": "_index", "value": "logs-unparsed" } }
  ]
}

5. 日期解析与数据清洗

一句话总结： 日志与业务数据进入 ES 前需要统一的日期、文本与字段格式，date/grok/dissect 处理器是清洗主力。

5.1 date 日期解析

原始时间字符串格式五花八门，date 处理器把它规范成 epoch 毫秒或 ISO 格式，落库后即可用 date_histogram 做时序分析：

{
  "processors": [
    {
      "date": {
        "field": "log_time",
        "formats": ["yyyy-MM-dd HH:mm:ss", "ISO8601"],
        "target_field": "@timestamp"
      }
    }
  ]
}

5.2 grok 与 dissect 文本解析

grok 用正则命名组从半结构化日志提取字段，dissect 用分隔符定位，速度更快但依赖固定结构：

{
  "processors": [
    { "grok": { "field": "message", "patterns": ["%{IP:client_ip} %{WORD:method} %{URIPATHPARAM:uri} %{NUMBER:status:int}"] } }
  ]
}

5.3 清洗组合拳

真实日志往往同时需要 trim、lowercase、convert 与 date。把幂等清洗放前面、解析放中间、结构化字段放后面，保证管道可重放：

{
  "processors": [
    { "trim": { "field": "message" } },
    { "gsub": { "field": "message", "pattern": "\\s+", "replacement": " " } },
    { "split": { "field": "user_agent", "separator": " " } }
  ]
}

6. 管道编排与调试

一句话总结： 管道可组合、可模拟、可版本化，simulate API 是无侵入调试管道的核心工具。

6.1 管道嵌套与复用

管道可以引用其他管道，形成分级处理；配合索引模板，管道随索引自动生效，采集端只需写入即可。全局管道（_ingest/pipeline）与每个索引的 default_pipeline 一起构成接入链。

6.2 simulate 模拟调试

_simulate 在真实索引上执行管道而不落库，可直接看到处理前后对比，是上线前必做的验证：

curl -X POST "localhost:9200/_ingest/pipeline/normalize-log/_simulate" -H "Content-Type: application/json" -d'
{
  "docs": [
    { "_source": { "msg": "  ERROR 500 /api/pay ", "response_time": "120" } }
  ]
}
'

6.3 版本与失败治理

给管道维护版本字段（如 pipeline_version），配合 on_failure 与死信索引实现可观测；测试文档用 _simulate 覆盖正常、缺失字段、类型异常三类样本，确保管道健壮。

7. 脚本性能与安全

一句话总结： 脚本默认被编译缓存、限时执行并在沙箱内运行，生产脚本要显式声明类型、避免高开销操作。

7.1 编译与缓存

Painless 脚本首次使用后编译缓存，默认 3000 条。带 params 的脚本可跨请求复用缓存；把常量写进源码会让每次变体都产生一次编译，应改用 params 传参。脚本执行有超时保护，避免死循环拖垮节点。

// 推荐：参数外置，缓存复用
"source": "return doc['price'].value * params.rate;",
"params": { "rate": 0.9 }

7.2 沙箱与白名单

Painless 运行在 SecurityManager 沙箱中，禁止访问类加载器、反射与文件系统。开启安全特性后，可通过 script.allowed_types 与 allowed_contexts 收紧可用范围，生产环境建议只保留 ingest 与 search 上下文。

7.3 性能红线

脚本读 doc 走列式存储，快于 _source；避免在脚本里做正则与字符串拼接热点；能用 script_score 与内置函数解决的不要手写循环。大批量重算场景（如全量 reindex）应评估改用显式字段而非每文档脚本。

8. 总结

Painless 与 Ingest 管道把「索引前数据加工」变成可声明、可测试、可观测的工程。优先用内置处理器，其次写脚本，最后才上 grok 正则；每个处理器都配 if 条件与 on_failure，让脏数据进得来、清得掉、看得见。脚本与管道的上游可阅读《数据建模与 Mapping 设计》与《ELK 日志分析体系》，集群侧治理参考《集群分片与高可用架构》。

延伸阅读

• 数据建模与 Mapping 设计
• ELK 日志分析体系：从采集到告警
• Query DSL 与相关性打分
• 聚合分析：从指标统计到多维下钻
• 集群分片与高可用架构