ElasticSearch如何聚合“copy_to”字段

Question

我需要按9个文件分组，并在ElasticSearch中获得每个组的计数，原始代码使用“脚本”，性能很差，所以我需要优化它。我设法创建了一个新的字段并使用“copy_to”，但是当我与新的字段聚合时，我发现了一些问题。

我使用“src ip”和“dst ip”字段作为测试，copy_to字段是“agg condition”。下面是映射：

PUT /test_index
{
  "settings": {
    "number_of_replicas": 0,
    "number_of_shards": 1
  },
  "mappings": {
      "dynamic_templates": [
    {
      "set_copy_to": {
        "match": "^(src|dst).+",
        "match_pattern": "regex",
        "mapping": {
          "copy_to": "aggCondition",
          "fields": {
            "keyword": {
              "ignore_above": 256,
              "type": "keyword"
            }
          },
          "type": "text"
        }
      }
    }
  ]
  }
}

然后我给它添加一些数据

{
  "srcIp":"192.0.0.1",
  "dstIp":"192.0.1.1"
}
{
  "srcIp":"192.0.1.1",
  "dstIp":"192.0.2.1"
}
{
  "srcIp":"192.0.2.1",
  "dstIp":"192.0.0.1"
}

然后我看到kibana中的映射，它看起来是这样的：

{
  "mappings": {
    "_doc": {
      "dynamic_templates": [
        {
          "set_copy_to": {
            "match": "^(src|dst).+",
            "match_pattern": "regex",
            "mapping": {
              "copy_to": "aggCondition",
              "fields": {
                "keyword": {
                  "ignore_above": 256,
                  "type": "keyword"
                }
              },
              "type": "text"
            }
          }
        }
      ],
      "properties": {
        "aggCondition": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        },
        "dstIp": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          },
          "copy_to": [
            "aggCondition"
          ]
        },
        "srcIp": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          },
          "copy_to": [
            "aggCondition"
          ]
        }
      }
    }
  }
}

GET /test_index/_search
{
  "aggs": {
    "Ips": {
      "terms": {
        "field": "aggCondition.keyword"
      }
    }
  }
}

  "aggregations" : {
    "Ips" : {
      "doc_count_error_upper_bound" : 0,
      "sum_other_doc_count" : 0,
      "buckets" : [
        {
          "key" : "192.0.0.1",
          "doc_count" : 2
        },
        {
          "key" : "192.0.1.1",
          "doc_count" : 2
        },
        {
          "key" : "192.0.2.1",
          "doc_count" : 2
        }
      ]
    }
  }

  "aggregations" : {
    "Ips" : {
      "doc_count_error_upper_bound" : 0,
      "sum_other_doc_count" : 0,
      "buckets" : [
        {
          "key" : "[192.0.0.1 192.0.1.1]",
          "doc_count" : 1
        },
        {
          "key" : "[192.0.1.1 192.0.2.1]",
          "doc_count" : 1
        },
        {
          "key" : "[192.0.2.1 192.0.0.1]",
          "doc_count" : 1
        }
      ]
    }
  }

我可以做什么来得到我期望的结果，或者有没有其他方法来有效地聚合多个字段？

Answer 1

dynamic_templates和copy_to不适合您的情况。最好动态计算一个索引SRC/DST IP对的新字段。您可以使用Ingest管道和append和join处理器来创建新字段。

PUT _ingest/pipeline/ip-pipeline
{
  "processors": [
    {
      "append": {
        "field": "srcDst",
        "value": ["{{{srcIp}}}", "{{{dstIp}}}"]
      }
    },
    {
      "join": {
        "field": "srcDst",
        "separator": "-"
      }
    }
  ]
}

然后，当您索引一个新文档时，您可以指定这个管道，新字段将被创建：

PUT my-index/_doc/1?pipeline=ip-pipeline
{
  "srcIp":"192.0.0.1",
  "dstIp":"192.0.1.1"
}

您的索引文档如下所示：

{
  "srcIp":"192.0.0.1",
  "dstIp":"192.0.1.1",
  "srcDst": "192.0.0.1-192.0.1.1"
}