Pipelines使用

pipeline 可让在建立索引之前对数据执行常见转换。例如可以使用管道删除字段、从文本中提取值以及丰富数据。管道由一系列的 Processor 组成，每个处理器按顺序运行，对传入文档进行特定更改。处理器运行后，Elasticsearch 将转换后的文档添加到数据流或索引中。

Pipelines使用

在Kibana的Stack Management > Ingest Node Pipelines路径下可以

● 查看管道列表并深入了解详细信息

● 编辑或克隆现有管道

● 删除管道

1. 创建新管道，请单击Create Pipeline

2. 为Pipeline添加Processor(如果希望使用JSON则点击import processor)，选择希望使用的Processor比如Script，在Source中添加处理规则将long类型的time除1000

3. 保存后再添加Date Processor，如图将UNIX格式的long类型time字段转换为Date类型，在target_field定义转换后的目标字段，默认是@timestamp

4. 最终Pipeline如下，可以点击箭头调整processor顺序

5. 为test1索引配置Pipeline,插入数据后可以看到数据经过处理

PUT test1
{
  "settings": {
    "index.default_pipeline":"trans"
  }
}

PUT test1/_doc/1
{
  "time":1635510843000
}

GET test1/_search

Pipeline API使用

使用方式:使用pipeline对每条写入ES的数据都添加写入时间。

注意:pipeline会对每条进入集群的数据进行处理，消耗更多写入性能

创建添加@timestamp的管道

PUT _ingest/pipeline/my_timestamp_pipeline
{
  "description": "Adds a field to a document with the time of ingestion",
  "processors": [
    {
      "set": {
        "field": "@timestamp",
        "value": "{{_ingest.timestamp}}"
      }
    }
  ]
}

为索引添加管道

PUT my_index
{
  "settings": {
    "default_pipeline": "my_timestamp_pipeline"
  }
}

我正在参与2023腾讯技术创作特训营第三期有奖征文，组队打卡瓜分大奖！