【Elasticsearch】Nested嵌套结构数据操作及聚合查询

明月AI

发布于 2022-02-23 14:48:21

6K0

发布于 2022-02-23 14:48:21

文章被收录于专栏：野生AI架构师

说明：本文需要一定的ES基础，以下基于ES6.8的版本。

ES的Nested数据类型允许我们存储一对多的数据，例如一个文章可以对应多个评论等，在正式开始之前，我们先生成一个用于测试的索引：

PUT /test_article
{
  "mappings": {
    "test_article": {
      "properties": {
        "id": {
          "type": "keyword"
        },
        "title": {
          "type": "text"
        },
        "tags": {
          "type": "text",
          "analyzer": "whitespace"
        },
        "data": {
          "type": "nested",    # 注意要指定type值
          "properties": {
            "system_type": {
              "type": "integer"
            },
            "affections": {
              "type": "keyword"
            },
            "themes": {
              "type": "text",
              "analyzer": "whitespace"
            }
          }
        }
      }
    }
  }
}

这是一个简化的文章表，data字段就是一个nested嵌套类型，存储不同平台（system_type）的标注数据（在一个文章内，system_type的值是唯一的），如倾向性（affections）、主题（themes）等。如果需要，nested类型是可以进行嵌套的。

然后插入一些测试数据：

POST /test_article/test_article/1
{
  "id": "1",
  "title": "标题1",
  "tags": "tag1 tag2 tag3",
  "data": []
}

POST /test_article/test_article/2
{
  "id": "2",
  "title": "标题2",
  "tags": "tag1 tag2 tag3",
  "data": [
    {
      "system_type": 1,
      "affections": "正面",
      "themes": "1 2"
    },
    {
      "system_type": 2,
      "affections": "中性",
      "themes": "1"
    }
  ]
}

POST /test_article/test_article
{
  "id": "3",
  "title": "标题4",
  "tags": "tag1 tag3",
  "data": [
    {
      "system_type": 1,
      "affections": "中性",
      "themes": "1 2"
    },
    {
      "system_type": 2,
      "affections": "负面",
      "themes": "1"
    }
  ]
}

POST /test_article/test_article
{
  "id": "5",
  "title": "标题5",
  "tags": "tag1 tag3",
  "data": [
    {
      "system_type": 2,
      "affections": "正面",
      "themes": "3 1"
    }
  ]
}

POST /test_article/test_article
{
  "id": "6",
  "title": "标题6",
  "tags": "tag2",
  "data": []
}

01 删除数据

这是比较简单的：

POST /test_article/test_article/2/_update
{
  "script": {
    "source": """
    ctx._source.data.removeIf(item -> item.system_type == 4)
    """
  }
}

使用脚本删除满足特定条件的数据，主要就是removeIf函数，该函数的参数应该是一个匿名函数（比较接近JS的匿名函数写法，就是一个语法糖），表示成python大概是这样：

lambda item: item.system_type == 4

item就是data中的元素，removeIf会把每个item都调用该匿名函数，如果得到true值就删除该元素。

02 修改数据

修改数据应该先判断数据是否已经存在：

POST /test_article/test_article/2/_update
{
  "script": {
    "source": """
    if (ctx._source.data != null) {
      for(e in ctx._source.data) {
        if (e.system_type == 2) {
          e.affections = "正面"; 
        }
      }
    }
    """
  }
}

上面的语句会删除data数据里，system_type值为2的记录。

修改数据成功之后，数据的版本号（_version）就会加1。

03 增加数据

增加数据的时候，先判断数据是否已经存在，不存在才执行增加，如果已经存在了，则执行修改：

POST /test_article/test_article/2/_update
{
  "script": {
    "source": """
    def is_in = false;
    if (ctx._source.data == null) {
      List ls = new ArrayList();
      ls.add(params.article);
    } else {
      for(e in ctx._source.data) {
        if (e.system_type == params.article.system_type) {
          is_in = true;
          for (String key: params.article.keySet()) {
            if (key != "system_type") {
              e[key] = params.article[key];
            }
          }
          break;
        }
      }
      if (is_in == false) {
        ctx._source.data.add(params.article);
      }
    }
    """,
    "params": {
      "article": {
        "system_type": 3,
        "affections": "负面",
        "themes": "3 2"
      }
    },
    "lang": "painless"
  }
}

这里比较特别的语法是：for (String key: params.article.keySet())

找了半天才发现对象可以使用keySet方法来获取key值，类似python中的dict.keys()。

另外，脚本中有参数需要使用的时候，比较好的实现应该是通过params进行传递，而不是硬编码到脚本中。

04 查询

nested数据的查询跟普通的查询有点不一样：

GET /test_article/_search
{
  "query": {
    "nested": {
      "path": "data",
      "query": {
        "term": {
          "data.system_type": 1
        }
      }
    }
  }
}

使用使用nested，并指定对应的path。但是要注意，这个查询只会对外层的记录进行过滤，并不会对nested内部的数据进行过滤。例如对于"data.system_type": 1，则data字段里有一条记录满足这个条件的，这个文章就会整体返回（当然可以通过_source命令进行筛选）。

如果说只想得到命中的nested数据，则可以使用inner_hits：

GET /test_article/_search
{
  "query": {
    "nested": {
      "path": "data",
      "query": {
        "bool": {
          "must": [
            {
              "term": {
                "data.system_type": {
                  "value": 2
                }
              }
            }
          ]
        }
      },
      "inner_hits": {}    # 返回满足条件的查询
    }
  },
  "size": 10
}

这时返回数据里就会增加一个inner_hits的字段：

{
  "hits" : {
    "total" : 3,
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "test_article",
        "_type" : "test_article",
        "_id" : "aqjGXH4BZeFFYagKZU_i",
        "_score" : 1.0,
        "_source" : {
          "id" : "5",
          "title" : "标题5",
          "tags" : "tag1 tag3",
          "data" : [       # 这里可以使用_source命令进行过滤掉
            {
              "system_type" : 2,
              "affections" : "正面",
              "themes" : "3 1"
            }, ......
          ]
        },
        "inner_hits" : {   # 这里只会返回命中的记录
          "data" : {
            "hits" : {
              "total" : 1,
              "max_score" : 1.0,
              "hits" : [
                {
                  "_index" : "test_article",
                  "_type" : "test_article",
                  "_id" : "aqjGXH4BZeFFYagKZU_i",
                  "_nested" : {
                    "field" : "data",
                    "offset" : 0
                  },
                  "_score" : 1.0,
                  "_source" : {
                    "system_type" : 2,
                    "affections" : "正面",
                    "themes" : "3 1"
                  }
                }
              ]
            }
          }
        }
      },
      ......
    ]
  }
}

05 聚合统计

在我们的场景中，场景的一个需要是，统计某个平台（system_type）下文章的倾向性的分布情况。开始的实现是这样：

GET /test_article/_search
{
  "size": 0,
  "aggs": {
    "positive": {
      "filter": {
        "nested": {
          "path": "data",
          "query": {
            "bool": {
              "must": [
                {
                  "term": {
                    "data.system_type": 2
                  }
                },
                {
                  "term": {
                    "data.affections": "正面"
                  }
                }
              ]
            }
          }
        }
      }
    },
    "negative": {
      "filter": {
        "nested": {
          "path": "data",
          "query": {
            "bool": {
              "must": [
                {
                  "term": {
                    "data.system_type": 2
                  }
                },
                {
                  "term": {
                    "data.affections": "负面"
                  }
                }
              ]
            }
          }
        }
      }
    },
    "neutral": {
      "filter": {
        "nested": {
          "path": "data",
          "query": {
            "bool": {
              "must": [
                {
                  "term": {
                    "data.system_type": 2
                  }
                },
                {
                  "term": {
                    "data.affections": "中性"
                  }
                }
              ]
            }
          }
        }
      }
    },
    "sensitive": {
      "filter": {
        "nested": {
          "path": "data",
          "query": {
            "bool": {
              "must": [
                {
                  "term": {
                    "data.system_type": 2
                  }
                },
                {
                  "term": {
                    "data.affections": "敏感"
                  }
                }
              ]
            }
          }
        }
      }
    }
  }
}

上面的语句是可以工作的，但是很罗嗦，差不多有100行，很多重复的代码，现在倾向性只有4个还勉强可以，如果有10个呢，那就这个语句就有两三百行。。。

于是优化成这样：

GET /test_article/_search
{
  "size": 0,
  "aggs": {
    "name": {
      "nested": {
        "path": "data"
      },
      "aggs": {
        "system_type_value": {
          "terms": {
            "field": "data.system_type"
          },
          "aggs": {
            "affections_value": {
              "terms": {
                "field": "data.affections"
              }
            }
          }
        }
      }
    }
  }
}

思路是先按data.system_type进行分桶，然后再按data.affections进行分桶，简洁了很多，但是这样的弊端是，我们本来只想统计某个平台下的数据，这里却会把所有平台的数据都进行统计了，浪费资源。

再优化：

GET /test_article/_search
{
  "size": 0, 
  "aggs": {
    "nested_data": {
      "nested": {
        "path": "data"
      },
      "aggs": {
        "filter_data": {
          "filter": {
            "term": {
              "data.system_type": 2
            }
          },
          "aggs": {
            "affections_value": {
              "terms": {
                "field": "data.affections"
              }
            }
          }
        }
      }
    }
  }
}

聚合里有一个filter的类型，之前居然没有注意到。通过filter过滤出满足条件的数据，再对data.affections进行分桶，完美解决。