前言

elasticsearch5.x 新增一个比较重要的特性 IngestNode。
之前如果需要对数据进行加工，都是在索引之前进行处理，比如logstash可以对日志进行结构化和转换，现在直接在es就可以处理了。
目前es提供了一些常用的诸如convert、grok之类的处理器，在使用的时候，先定义一个pipeline管道，里面设置文档的加工逻辑，在建索引的时候指定pipeline名称，那么这个索引就会按照预先定义好的pipeline来处理了。

Ingest Attachment Processor Plugin

处理文档附件，替换之前的 mapper attachment plugin。
默认存储附件内容必须base64编码的数据，不想base64转换，可以使用CBOR（没有试验）
官网说明：

The source field must be a base64 encoded binary. 
If you do not want to incur the overhead of converting back and forth between base64, 
you can use the CBOR format instead of JSON and specify the field as a bytes array instead of a string representation. 
The processor will skip the base64 decoding then.

安装

./bin/elasticsearch-plugin install ingest-attachment

卸载

./bin/elasticsearch-plugin remove ingest-attachment

用管道处理单个附件示例(Using the Attachment Processor in a Pipeline)

1.创建管道single_attachment

PUT _ingest/pipeline/single_attachment
{
  "description" : "Extract single attachment information",
  "processors" : [
    {
      "attachment" : {
        "field": "data",
        "indexed_chars" : -1,
        "ignore_missing" : true
      }
    }
  ]
}

2.创建index

PUT /index1
{
    "mappings" : {
        "type1" : {
            "properties" : {
                "id": {
                    "type": "keyword"
                },
                "filename": {
                    "type": "text",
                    "analyzer": "english"
                },
                "data":{
                    "type": "text",
                    "analyzer": "english"
                }
            }
        }
    }
}

3.索引数据

PUT index1/type1/1?pipeline=single_attachment&refresh=true&pretty=1
{
    "id": "1",
    "filename": "1.txt",
    "data" : "e1xydGYxXGFuc2kNCkxvcmVtIGlwc3VtIGRvbG9yIHNpdCBhbWV0DQpccGFyIH0="
}

PUT index1/type1/2?pipeline=single_attachment&refresh=true&pretty=1
{
  "id": "2",
  "subject": "2.txt",
  "data": "dGVzdGluZyBteSBmaXJzdCBlbmNvZGVkIHRleHQ="
}

4.查看结果

GET index1/type1/1
GET index1/type1/2

POST index1/type1/_search?pretty=true
{
  "query": {
    "match": {
      "attachment.content_type": "text plain"
    }
  }
}

POST index1/type1/_search?pretty=true
{
  "query": {
    "match": {
      "attachment.content": "testing"
    }
  },
  "highlight": {
    "fields": {
      "attachment.content": {}
    }
  }
}

返回结果

"hits": [
    {
        "_index": "index1",
        "_type": "type1",
        "_id": "2",
        "_score": 0.2824934,
        "_source": {
            "data": "dGVzdGluZyBteSBmaXJzdCBlbmNvZGVkIHRleHQ=",
            "attachment": {
                "content_type": "text/plain; charset=ISO-8859-1",
                "language": "et",
                "content": "testing my first encoded text",
                "content_length": 30
            },
            "subject": "2.txt",
            "id": "2"
        },
        "highlight": {
            "attachment.content": [
                "<em>testing</em> my first encoded text"
            ]
        }
    }
]

用管道处理多个附件示例(Using the Attachment Processor with arrays)、

1.创建管道multi_attachment

PUT _ingest/pipeline/multi_attachment
{
  "description" : "Extract attachment information from arrays",
  "processors" : [
    {
      "foreach": {
        "field": "attachments",
        "processor": {
          "attachment": {
            "target_field": "_ingest._value.attachment",
            "field": "_ingest._value.data",
            "indexed_chars" : -1,
            "ignore_missing" : true
          }
        }
      }
    }
  ]
}