ElasticSearch 如何聚合“copy_to”字段

Question

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

我使用 'srcIp' 和 'dstIp' 字段作为测试，copy_to 字段是 'aggCondition'。这是映射：

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"
          ]
        }
      }
    }
  }
}

然后我聚合使用新字段'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 对的新字段。您可以使用带有append 和join 处理器的ingest pipeline 来创建新字段。

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"
}

然后您可以在新的 srcDst 字段上运行聚合查询并获得您期望的结果。