Elastic Ingest Pipeline 拆分字段并创建嵌套字段

Question

亲爱的好帮手，

我有一个通过 Kafka 由数据库提供的索引。现在这个数据库有一个字段，它聚合了一些信息，比如键/值；核心价值; （不要问原因，我不知道是谁设计的，为什么喜欢它；-)）

93/4; 34/12；

它可以为空，也可以包含 1..n 个键/值对。

我想使用摄取管道，理想情况下有一个“嵌套”字段，其中包含该字段中的所有值。

大概是这样的：

{“类别”： {“93”：7， “82”：4 } }

用例如下：我们希望将这些类别的过滤数量的总和可视化（它们告诉我特定过程花费了多少分钟）并将它们关联到范围内。

示例：我过滤 x、y、z 类别，然后将当天没有延迟的文档分组，其中延迟最多 5 分钟，延迟在 5 到 15 分钟之间。

我试图用 kv 处理器将字段整齐地分开，并想从那里开始工作，但我猜这是一个完全错误的方法。

"kv": {
      "field": "IncomingField",
      "field_split": ";",
      "value_split": "/",
      "target_field": "delays",
      "ignore_missing": true,
      "trim_key": "\\s",
      "trim_value": "\\s",
      "ignore_failure": true
    }

当我测试管道时，它似乎没问题

"delays": {
            "62": "3",
            "86": "2"
          }

但有两件事不起作用。

1. 我无法预先知道我有多少这些组合，因此在同一管道中转换字符串 t int 的值是一个问题。
2. 当我想创建一个 kibana 索引模式时，我最终会得到许多字段，例如 delay.82 和 delay.82.keyword 这对于用例来说根本没有意义，因为我无法过滤（仅获取键是 x,y,z) 和聚合之一的延迟。

我已经研究过其他处理器 (dorexpander)，但无法真正了解如何使其正常工作。

我希望我的问题很清楚（我缺乏英语技能，抱歉），并且希望有人能指出我正确的方向。

非常感谢！

Answer 1

您应该将它们构造为具有共享访问器的对象数组，例如：

[ {key: 93, value: 7}, ...]

这样，您就可以在categories.key 和categories.value 上进行聚合。

这意味着使用自定义脚本处理器迭代类别的entrySet()，如下所示：

POST _ingest/pipeline/_simulate
{
  "pipeline": {
    "description": "extracts k/v pairs",
    "processors": [
      {
        "script": {
          "source": """
            def categories = ctx.categories;
            def kv_pairs = new ArrayList();
    
            for (def pair : categories.entrySet()) {
              def k = pair.getKey();
              def v = pair.getValue();
              kv_pairs.add(["key": k, "value": v]);
            }
            
            ctx.categories = kv_pairs;
          """
        }
      }
    ]
  },
  "docs": [
    {
      "_source": {
        "categories": {
          "82": 4,
          "93": 7
        }
      }
    }
  ]
}

P.S.：请确保您的 categories 字段是 mapped as nested b/c 否则您将失去键和值之间的连接（也称为展平）。