数据清洗-使用es的ingest

 通常es产品里面，数据清洗的logstash，这里使用另外的ingest做简单的数据处理

这里是根据逗号分隔数据，变成数组

PUT spring_blogs/_doc/1
{
  "title":"Introducing spring framework......",
  "tags":"spring,spring boot,spring cloud",
  "content":"You konw, for spring framework"
}

POST _ingest/pipeline/_simulate
{
  "pipeline": {
    "description": "to split blog tags",
    "processors": [
      {
        "split": {
          "field": "tags",
          "separator": ","
        }
      }
    ]
  },
  "docs": [
    {
      "_index": "index",
      "_id": "id",
      "_source": {
        "title": "Introducing spring framework......",
        "tags": "spring,spring boot,spring cloud",
        "content": "You konw, for spring framework"
      }
    },
    {
      "_index": "index",
      "_id": "idxx",
      "_source": {
        "title": "Introducing cloud computering",
        "tags": "docker,k8s,ingrest",
        "content": "You konw, for cloud"
      }
    }
  ]
}

# 为ES添加一个 Pipeline
PUT _ingest/pipeline/spring_blog_pipeline
{
  "description": "a spring blog pipeline",
  "processors": [
      {
        "split": {
          "field": "tags",
          "separator": ","
        }
      },

      {
        "set":{
          "field": "views",
          "value": 0
        }
      }
    ]
}

#查看Pipleline
GET _ingest/pipeline/spring_blog_pipeline

#测试pipeline
POST _ingest/pipeline/spring_blog_pipeline/_simulate
{
  "docs": [
    {
      "_source": {
        "title": "Introducing cloud computering",
        "tags": "docker,k8s,ingrest",
        "content": "You konw, for cloud"
      }
    }
  ]
}

DELETE spring_blogs

PUT spring_blogs/_doc/1
{
  "title":"Introducing spring framework......",
  "tags":"spring,spring boot,spring cloud",
  "content":"You konw, for spring framework"
}

#使用pipeline更新数据
PUT spring_blogs/_doc/2?pipeline=spring_blog_pipeline
{
  "title": "Introducing cloud computering",
  "tags": "docker,k8s,ingrest",
  "content": "You konw, for cloud"
}

POST spring_blogs/_search


#增加update_by_query的条件
POST spring_blogs/_update_by_query?pipeline=spring_blog_pipeline
{
    "query": {
        "bool": {
            "must_not": {
                "exists": {
                    "field": "views"
                }
            }
        }
    }
}

最终处理后的数据

{
  "took" : 0,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 2,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "spring_blogs",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "title" : "Introducing cloud computering",
          "content" : "You konw, for cloud",
          "views" : 0,
          "tags" : [
            "docker",
            "k8s",
            "ingrest"
          ]
        }
      },
      {
        "_index" : "spring_blogs",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "title" : "Introducing spring framework......",
          "content" : "You konw, for spring framework",
          "views" : 0,
          "tags" : [
            "spring",
            "spring boot",
            "spring cloud"
          ]
        }
      }
    ]
  }
}

还可以使用Script Prcessor，这种自由度更大，可以处理稍微复杂点数据

POST _ingest/pipeline/_simulate
{
  "pipeline": {
    "description": "to split spring blog tags",
    "processors": [
      {
        "split": {
          "field": "tags",
          "separator": ","
        }
      },
      {
        "script": {
          "source": """
          if(ctx.containsKey("title")){
            ctx.content_length = ctx.title.length();
          }else{
            ctx.content_length=0;
          }


          """
        }
      },
      {
        "set": {
          "field": "views",
          "value": 0
        }
      }
    ]
  },
  "docs": [
    {
      "_index": "index",
      "_id": "id",
      "_source": {
        "title": "Introducing spring framework......",
        "tags": "spring,spring boot,spring cloud",
        "content": "You konw, for spring framework"
      }
    },
    {
      "_index": "index",
      "_id": "idxx",
      "_source": {
        "title": "Introducing cloud computering",
        "tags": "docker,k8s,ingrest",
        "content": "You konw, for cloud"
      }
    }
  ]
}