如何将OpenSearch索引中字符串类型的message字段转为JSON

我有一个OpenSearch索引，其中包含一个字符串类型的message.message字段，字段定义如下：

{"name":"message.message","type":"string","esTypes":["text"],"count":0,"scripted":false,"searchable":true,"aggregatable":false,"readFromDocValues":false}

示例数据：

"_source" : {
          "message" : {
            "message" : "user: AB, from: home, to: /app1"
          }
        }

希望将该message.message字段转换为结构化JSON格式，以便单独访问message.user、message.from和message.to的值。

• 更新索引映射
  首先需要修改索引的映射，为message对象添加结构化的子字段（user、from、to），建议使用keyword类型（方便精确查询和聚合）：

  PUT /your-index-name/_mapping
  {
    "properties": {
      "message": {
        "properties": {
          "message": {"type": "text"},
          "user": {"type": "keyword"},
          "from": {"type": "keyword"},
          "to": {"type": "keyword"}
        }
      }
    }
  }
  

  替换your-index-name为你的实际索引名称。

• 创建Ingest Pipeline
  创建一个用于解析message.message字符串的管道，使用dissect处理器（适合固定格式的字符串解析，性能比grok更高）：

  PUT _ingest/pipeline/parse-message-pipeline
  {
    "processors": [
      {
        "dissect": {
          "field": "message.message",
          "pattern": "user: %{message.user}, from: %{message.from}, to: %{message.to}"
        }
      }
    ]
  }
  

  这个处理器会按照指定的模式，把字符串中的值提取到message.user、message.from、message.to字段中。

• 批量更新已有数据
  使用_update_by_query命令，对索引中所有文档应用上述管道，完成已有数据的结构化转换：

  POST /your-index-name/_update_by_query?pipeline=parse-message-pipeline
  

  这个操作会遍历索引中的所有文档，解析message.message并填充结构化字段。

• 处理新写入的数据
  如果需要让后续写入的自动完成解析，可以将管道设置为索引的默认管道：

  PUT /your-index-name/_settings
  {
    "index.default_pipeline": "parse-message-pipeline"
  }
  

  之后新写入的文档会自动经过管道解析，生成结构化字段。

  若不想设置默认管道，也可以在写入文档时显式指定管道：

  POST /your-index-name/_doc?pipeline=parse-message-pipeline
  {
    "message": {
      "message": "user: CD, from: office, to: /app2"
    }
  }
  

内容的提问来源于stack exchange，提问作者apprunner2186