Elasticsearch 合并文档作为响应

Question

我有 3 个索引中的数据。我想使用来自其他索引的信息生成发票报告。例如下面是每个索引的示例文档
用户索引

{
    "_id": "userId1",
    "name": "John"
}

发票索引

{
    "_id": "invoiceId1",
    "userId": "userId1",
    "cost": "10000",
    "startdate": "",
    "enddate": ""
}

订单指数

{
    "_id": "orderId1",
    "userId": "userId1",
    "productName": "Mobile"
}

我想通过结合这三个索引的信息来生成发票报告如下

{
    "_id": "invoiceId1",
    "userName": "John",
    "productName": "Mobile",
    "cost": "10000",
    "startdate": "",
    "enddate": ""
}

如何编写通过组合来自其他索引文档的信息返回响应的 Elasticsearch 查询？

Answer 1

您不能 在 Elasticsearch 中执行查询时连接，并且需要对数据进行非规范化以有效地检索和分组数据。

话虽如此，您可以:

杠杆multi-target syntax并一次查询多个索引

使用 OR查询id和 userId -- 因为其中任何一个在您的任何文档中至少被引用一次

然后通过名为 scripted metric aggregations 的 map/reduce 工具简单地连接您的数据

快速旁注:您将无法使用 _id文档中的关键字，因为它是保留的。
假设您的文档和索引的结构如下:

POST users_index/_doc
{"id":"userId1","name":"John"}

POST invoices_index/_doc
{"id":"invoiceId1","userId":"userId1","cost":"10000","startdate":"","enddate":""}

POST orders_index/_doc
{"id":"orderId1","userId":"userId1","productName":"Mobile"}

以下是脚本化指标聚合的样子:

POST users_index,invoices_index,orders_index/_search
{
  "size": 0, 
  "query": {
    "bool": {
      "should": [
        {
          "term": {
            "id.keyword": {
              "value": "userId1"
            }
          }
        },
        {
          "term": {
            "userId.keyword": {
              "value": "userId1"
            }
          }
        }
      ]
    }
  },
  "aggs": {
    "group_by_invoiceId": {
      "scripted_metric": {
        "init_script": "state.users = []; state.invoices = []; state.orders = []",
        "map_script": """
          def source = params._source;
          
          if (source.containsKey("name")) {
            // we're dealing with the users index
            state.users.add(source);
          } else if (source.containsKey("cost")) {
            // we're dealing with the invoices index
            state.invoices.add(source);
          } else if (source.containsKey("productName")) {
            // we're dealing with the orders index
            state.orders.add(source);
          }
        """,
        "combine_script": """
          def non_empty_state = [:];
          for (entry in state.entrySet()) {
            if (entry != null && entry.getValue().length > 0) {
              non_empty_state[entry.getKey()] = entry.getValue();
            }
          }
          return non_empty_state;
        """,
        "reduce_script": """
          def final_invoices = [];
          
          def all_users = [];
          def all_invoices = [];
          def all_orders = [];
          
          // flatten all resources
          for (state in states) {
            for (kind_entry in state.entrySet()) {
              def map_kind = kind_entry.getKey();
              if (map_kind == "users") {
                all_users.addAll(kind_entry.getValue());
              } else if (map_kind == "invoices") {
                all_invoices.addAll(kind_entry.getValue());
              } else if (map_kind == "orders") {
                all_orders.addAll(kind_entry.getValue());
              } 
            }
          }
          
          // iterate the invoices and enrich them
          for (invoice_entry in all_invoices) {
            def invoiceId = invoice_entry.id;
            def userId = invoice_entry.userId;
            def userName = all_users.stream().filter(u -> u.id == userId).findFirst().get().name;
            def productName = all_orders.stream().filter(o -> o.userId == userId).findFirst().get().productName;
            def cost = invoice_entry.cost;
            def startdate = invoice_entry.startdate;
            def enddate = invoice_entry.enddate;
            
            final_invoices.add([
              'id': invoiceId,
              'userName': userName,
              'productName': productName,
              'cost': cost,
              'startdate': startdate,
              'enddate': enddate
            ]);
          }
          
          return final_invoices;
        """
      }
    }
  }
}

哪个会回来

{
  ...
  "aggregations" : {
    "group_by_invoiceId" : {
      "value" : [
        {
          "cost" : "10000",
          "enddate" : "",
          "id" : "invoiceId1",
          "userName" : "John",
          "startdate" : "",
          "productName" : "Mobile"
        }
      ]
    }
  }
}

总而言之，有一些解决方法可以实现查询时连接。同时，不应该在生产中使用这样的脚本，因为它们可能需要很长时间。
相反，在查询解析并返回特定于索引的命中后，应在 Elasticsearch 之外模拟此聚合。
顺便说一句——我设置了 size: 0仅返回聚合结果，因此如果您想获得一些实际点击，请增加此参数。