Elasticsearch Ingest Pipeline数据预

1、Pipeline定义

在Elasticsearch 5.0版本以后引入了ingest node，用于在文档被索引之前进行预处理。
Pipeline 定义了一系列按顺序执行的 processors, 一个 pipeline 由 description 和 processors两部分组成：

PUT _ingest/pipeline/my-pipeline-id
{
  "description" : "...",
  "processors" : [ ... ]
}

2、Pipeline获取文档信息

2.1 pipeline中的 processors 可以获取到传入的文档的原始数据以及元数据，并进行读写操作。获取单个字段可以直接使用字段名：

{
  "set": {
    "field": "my_field",
    "value": 582.1
  }
}

或者使用_source前缀：

{
  "set": {
    "field": "_source.my_field",
    "value": 582.1
  }
}

2.2 pipeline同样可以访问文档中的元信息，_index、_type、_id、_routing等。

{
  "set": {
    "field": "_id",
    "value": "1"
  }
}

2.3 pipeline中还可以获取Ingest 的元信息，ingest的元数据信息是只存在与文档预处理期间，当文档被pipeline处理完成，ingest元信息也就消息了。

{
  "set": {
    "field": "received",
    "value": "{{_ingest.timestamp}}"
  }
}

此处使用了{{}},这种方式同样可以作为模板获取文档内的字段，比如下面field_c的内容是field_a与field_b连接。

{
  "set": {
    "field": "field_c",
    "value": "{{field_a}} {{field_b}}"
  }
}

在pipeline中还支持使用动态字段，如下将service字段的值当做字段名，将code的值当做字段值：

{
  "set": {
    "field": "{{service}}",
    "value": "{{code}}"
  }
}

3、使用Simulate Pipeline API

在定义了pipeline之后，可以使用Simulate Pipeline API 对文档进行预处理执行测试。

POST /_ingest/pipeline/my-pipeline-id/_simulate
{
  "docs": [
    {
      "_index": "index",
      "_id": "id",
      "_source": {
        "foo": "bar"
      }
    },
    {
      "_index": "index",
      "_id": "id",
      "_source": {
        "foo": "rab"
      }
    }
  ]
}

POST /_ingest/pipeline/_simulate
{
  "pipeline" :
  {
    "description": "_description",
    "processors": [
      {
        "set" : {
          "field" : "field2",
          "value" : "_value"
        }
      }
    ]
  },
  "docs": [
    {
      "_index": "index",
      "_id": "id",
      "_source": {
        "foo": "bar"
      }
    },
    {
      "_index": "index",
      "_id": "id",
      "_source": {
        "foo": "rab"
      }
    }
  ]
}

4、常用processors类型详解

具体详解请看官方文档：https://www.elastic.co/guide/en/elasticsearch/reference/7.9/ingest-processors.html
此处主要说明几个关键常用。
4.1 Append Processor
该Processor追加一个或者多个值到现存的数组字段中，如果字段本身是标量则将其转换成数组再添加，如果字段不存在则创建包含该值的数字。

{
  "append": {
    "field": "tags",
    "value": ["production", "{{app}}", "{{owner}}"]
  }
}

4.2 Convert Processor
该Processor可以转换当前正在处理的字段数据类型，比如将一个String转换成Integer,如果是数组，则数组中所有成员都会被转换。目前只支持：integer, long, float, double, string, boolean, and auto。

PUT _ingest/pipeline/my-pipeline-id
{
  "description": "converts the content of the price field to an double",
  "processors" : [
    {
      "convert" : {
        "field" : "price",
        "type": "double"
      }
    }
  ]
}

4.3 Date Index Name Processor
该 Processor 可以指定文档中的日期字段，将文档分配到符合指定时间格式的索引中，前提是按照官方提供的说明来进行使用。

PUT _ingest/pipeline/monthlyindex
{
  "description": "monthly date-time index naming",
  "processors" : [
    {
      "date_index_name" : {
        "field" : "date1",
        "index_name_prefix" : "my-index-",
        "date_rounding" : "M"
      }
    }
  ]
}

PUT /my-index/_doc/1?pipeline=monthlyindex
{
  "date1" : "2020-04-25T12:02:01.789Z"
}

{
  "_index" : "my-index-2020-04-01",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 1,
  "result" : "created",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 55,
  "_primary_term" : 1
}

上面的不会放入myindex索引中，而是myindex-2016-04-01 索引中，该功能与template配合使用可以按日期进行索引。

4.4 Script Processor
该 Processor 是 Ingest 中功能最强大的Processor，利用Elasticsearch提供的脚本能力。

{
  "script": {
    "lang": "painless",
    "source": "ctx.field_a_plus_b_times_c = (ctx.field_a + ctx.field_b) * params.param_c",
    "params": {
      "param_c": 10
    }
  }
}

4.5 Set Processor
该 Processor 用于指定字段的值，如果该字段存在时，则修改字段的值，如果该字段不存在，则新增字段并设置该字段的值。

{
  "description" : "sets the value of count to 1"
  "set": {
    "field": "count",
    "value": 1
  }
}

该处理器也可以用其他字段的值动态赋值：

PUT _ingest/pipeline/set_os
{
  "description": "sets the value of host.os.name from the field os",
  "processors": [
    {
      "set": {
        "field": "host.os.name",
        "value": "{{os}}"
      }
    }
  ]
}

POST _ingest/pipeline/set_os/_simulate
{
  "docs": [
    {
      "_source": {
        "os": "Ubuntu"
      }
    }
  ]
}