Elasticsearch 中的嵌套查询？

Question

我的团队拥有多个仪表板，并且正在考虑迁移到 Elasticsearch 以整合软件堆栈的可能性。我们公开的一种常见图表类似于“每天结束时待处理的工作流程是什么？”。以下是一些示例数据：

day workflow_id version status
20151101    1   1   In Progress
20151101    2   1   In Progress
20151102    1   2   In Progress
20151102    3   1   In Progress
20151102    4   1   In Progress
20151102    2   2   Completed
20151103    1   3   Completed
20151103    3   2   In Progress
20151104    3   3   Completed
20151105    4   2   Completed

每当工作流程中发生更改时，都会插入一条新记录，这可能会也可能不会更改状态。具有 max(version) 的记录是 workflow_id 的最新数据。

我们的目标是制作一个图表来显示每天结束时“进行中”和“已完成”工作流的总数。这应该只考虑到当天版本号最大的记录。这可以通过嵌套查询在 SQL 中完成：

with 

snapshot_dates as 
(select distinct day from workflow),

snapshot as 
(select d.day, w.workflow_id, max(w.version) as max_version
from snapshot_dates d, workflow w
where d.day >= w.day
group by d.day, w.workflow_id
order by d.day, w.workflow_id)

select s.day, w.status, count(1) 
from workflow w join snapshot s on w.workflow_id=s.workflow_id and w.version = s.max_version
group by s.day, w.status
order by s.day, w.status;

这是查询的预期输出：

day,status,count  
20151101,In Progress,2  
20151102,Completed,1  
20151102,In Progress,3  
20151103,Completed,2  
20151103,In Progress,2  
20151104,Completed,3  
20151104,In Progress,1  
20151105,Completed,4 

我还是 Elasticsearch 的新手，想知道 Elasticsearch 是否可以在不使用应用程序端逻辑的情况下通过正确定义映射和查询来执行类似的查询。更一般地说，使用 Elasticsearch 解决此类问题的最佳实践是什么？

Answer 1

我尝试使用bucket selector aggregation 找到解决方案，但我被困在了某一点上。我在elasticsearch forum 中讨论过同样的问题。以下是Christian Dahlqvist的建议。

除此之外，您还可以将记录索引到以工作流为中心的 具有唯一标识符的索引，例如工作流 ID，作为文档 ID。 如果同一工作流程有多个更新，每个更新都会导致 将保留更新和最新状态。运行聚合 在此索引中查找当前或最新状态将是 由于您只有一个 每个工作流记录，不需要根据 与其他文档的关系。

因此，根据此建议，您应该在索引时使用Workflow Id 作为文档 ID。每当该工作流有更新时，您都会使用工作流 ID 更新新版本和日期。假设索引名称为workflow，索引类型为workflow_status。所以workflow_status 类型的映射如下：

{
    "workflow_status": {
        "properties": {
            "date": {
                "type": "date",
                "format": "strict_date_optional_time||epoch_millis"
            },
            "status": {
                "type": "string",
                "index": "not_analyzed"
            },
            "version": {
                "type": "long"
            },
            "workFlowId": {
                "type": "long"
            }
        }
    }
}

继续将文档添加/更新到此索引类型，并将 workFlowId 作为文档 ID。

现在，为了按天显示图表，您可能需要创建另一种索引类型，例如 per_day_workflow 并使用以下映射：

{
    "per_day_workflow": {
        "properties": {
            "date": {
                "type": "date",
                "format": "strict_date_optional_time||epoch_millis"
            },            
            "in_progress": {
                "type": "long"
            },
            "completed": {
                "type": "long"
            }
        }
    }
}

该索引将保存每天的数据。因此，您需要创建一个将在一天结束时运行的作业，并使用以下聚合搜索从 workflow_status 索引类型中获取总的“进行中”和“已完成”工作流：

POST http://localhost:9200/workflow/workflow_status/_search?search_type=count

    {
        "aggs": {
            "per_status": {
                "terms": {
                    "field": "status"
                }
            }
        }
    }

响应将如下所示（我在您的示例数据上运行日期为 2015-11-02）：

{
    "took": 3,
    "timed_out": false,
    "_shards": {
        "total": 5,
        "successful": 5,
        "failed": 0
    },
    "hits": {
        "total": 4,
        "max_score": 0,
        "hits": []
    },
    "aggregations": {
        "per_status": {
            "doc_count_error_upper_bound": 0,
            "sum_other_doc_count": 0,
            "buckets": [
                {
                    "key": "In Progress",
                    "doc_count": 3
                },
                {
                    "key": "Completed",
                    "doc_count": 1
                }
            ]
        }
    }
}

您需要从此响应中提取 In Progress 和 Completed 计数并将它们添加到带有今天日期的 per_day_workflow 索引类型。

现在，当您需要图表的每日数据时，您可以轻松地从此 per_day_workflow 索引类型中获取数据。