Elasticsearch嵌套查询映射的DynamoDB数据无返回结果求助

刚看到你的问题，这不就是我上个月踩过的一模一样的坑嘛！我当时也是把DynamoDB数据同步到ES后，明明索引里有目标文档，但最简嵌套查询啥都查不到，试了caller.S这种路径也没用，还连个报错都没有，折腾了好半天。

问题根源找到了！ #

后来反复排查才发现：Elasticsearch要正确识别并索引嵌套数据，必须要求嵌套字段是标准的{key: value}对象格式，但DynamoDB导出的数据（比如带S/N这类类型标记的结构），ES根本没法把它当成正常的嵌套对象处理——相当于你的嵌套字段其实是个特殊结构，不是ES认知里的嵌套对象，自然查不到。

解决方案：用预处理管道转数据 #

要搞定这个，得在数据进入ES之前，通过**Ingest Pipeline（预处理管道）**把DynamoDB的带类型包装的数据转换成标准对象：

1. 创建转换管道 #

比如针对caller这类字段，我们可以用脚本处理器把{S: "xxx"}这种DynamoDB格式转成直接的字符串值，如果是多层嵌套的结构，也可以递归处理：

PUT _ingest/pipeline/dynamodb-to-es-transform
{
  "description": "把DynamoDB的带类型结构转成ES能识别的标准对象",
  "processors": [
    {
      "script": {
        "source": """
          // 处理单个字符串类型字段，比如caller
          if (ctx.caller != null && ctx.caller.S != null) {
            ctx.caller = ctx.caller.S;
          }
          // 处理多层嵌套的Map类型（比如某个字段是DynamoDB的M类型）
          if (ctx.userInfo != null && ctx.userInfo.M != null) {
            ctx.userInfo = ctx.userInfo.M;
            // 遍历处理嵌套里的每个字段
            for (def entry : ctx.userInfo.entrySet()) {
              if (entry.value.S != null) {
                entry.value = entry.value.S;
              } else if (entry.value.N != null) {
                entry.value = Integer.parseInt(entry.value.N);
              }
            }
          }
        """
      }
    }
  ]
}

2. 把管道关联到索引 #

两种方式：

• 创建索引时指定默认管道，后续同步的数据自动走转换：

PUT /your-target-index
{
  "settings": {
    "index.default_pipeline": "dynamodb-to-es-transform"
  },
  "mappings": {
    "properties": {
      "caller": {
        "type": "text"
      },
      // 其他字段的映射定义
    }
  }
}

• 如果是用Lambda+DynamoDB Streams同步的话，在写入ES的请求里加上pipeline=dynamodb-to-es-transform参数，显式指定用这个管道。

3. 重新处理已有的旧数据 #

对于已经同步到ES里的历史数据，得重新索引来应用转换：

POST _reindex
{
  "source": {
    "index": "your-old-index"
  },
  "dest": {
    "index": "your-target-index",
    "pipeline": "dynamodb-to-es-transform"
  }
}

等这些都做完，你的嵌套字段就变成ES能识别的标准对象了，再用常规的嵌套查询就能正常返回结果啦！

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