0

我有一个弹性索引，其中包含用户状态历史记录的文档。数据看起来像这样；

  {
    "session_id": "yunus",
    "state_name": "start",
    "entry_time": "2016-11-09 15:27:03"
  },
  {
    "session_id": "yunus",
    "state_name": "end",
    "entry_time": "2016-11-09 16:30:00"
  },
  {
    "session_id": "can",
    "state_name": "start",
    "entry_time": "2016-11-09 12:01:00"
  },
  {
    "session_id": "rick",
    "state_name": "start",
    "entry_time": "2016-11-09 09:00:00"
  },
  {
    "session_id": "rick",
    "state_name": "end",
    "entry_time": "2016-11-10 10:00:00"
  }

我想按状态名称与日期直方图聚合，但只针对当时相关的最后一个状态。所以结果可以是；

2016-11-08 
start = 0
end = 0

2016-11-09 
start = 2
end = 1

2016-11-10 
start = 1
end = 2

实际上计划是生成带有时间线的分组条形图，以显示状态随时间的变化。

我尝试了几种方法，例如聚合管道、热门歌曲，但没有取得任何进展。

任何帮助表示赞赏。

4

1 回答 1

0

对于任何有兴趣的人，我用火花解决了它。我使用弹性火花从弹性搜索中读取，然后写回弹性搜索。

这是从esas中读取的内容Rdd；

val allData = sc.esRDD(s"states_${id}/log", query)

然后我首先按会话 id 分组，按日期排序以仅查找会话的最新状态；

val latestStates = allData.groupBy(k => k._2.get("session_id").get).map(k => (k._2).reduceLeft((d1, d2) => {
  d1._2.get("timestamp").get.asInstanceOf[Long] > d2._2.get("timestamp").get.asInstanceOf[Long] match {
    case true => d1
    case false => d2
  }
})).map(_._2)

一旦我有了最新的会话状态，我就会过滤退出状态，然后按值计数；

val stateSummary = latestStates
  .filter(s => s.isDefinedAt("state_id") && s("state_id").asInstanceOf[Long] != -1)
  .map(s => (s("state_id"), s("state_name")))
  .countByValue()
  .map(d => Map("state_id" -> d._1._1.asInstanceOf[Long], "state_name" -> d._1._2.asInstanceOf[String], "count" -> d._2)).toList

现在我们有当前状态的会话数。（当前是可配置的，因此我们可以将其设置为特定时间），只剩下一件事，写回elasticsearch；

sc.makeRDD(Seq(finalElasticDoc)).saveToEs(s"states_${id}/analytic_daily")
于 2016-11-28T08:03:44.893 回答