Построение динамических запросов агрегирования для asticsearch с использованием клиента astic4s

Question

Можно ли выполнять многоуровневые запросы агрегации динамически, используя DSL эластичного 4-го уровня.

Используя http-клиент это просто

multiLevelAggregation

 Вход: поля [0..N]
 Вывод: данные сгруппированы по кортежу полей

 шаги:
 1. Построить многоуровневый запрос агрегации эластичного поиска (JSON)
 2. Выполните запрос на сервере Flexiblesearch.
 3. Свести результат и вернуть 

Но как это сделать, используя astic4s или Java-клиент.

Answer 1

После тщательного понимания моей проблемы мне удалось найти решение этой проблемы. Сначала я подумал, что это будет ограничением эластичности, но это не так, его легко динамически создавать многопольные агрегационные запросы с помощью клиента astic4s. Вот мое решение.

//For building aggregation query
def buildAgg(groups: Seq[String])(leafAggBuilder: () => AbstractAggregationDefinition): AbstractAggregationDefinition = {
  groups match {
    case x :: xs => aggregation.terms("termAgg").field(x).aggregations(buildAgg(xs)(leafAggBuilder))
    case Nil => leafAggBuilder()
  }
}

//An example leaf aggregation builder
def buildLeafAgg(aggFuncInfo: Pair[String, String])(): AbstractAggregationDefinition = {
  aggFuncInfo._1 match {
    case "avg" => aggregation.avg("aggFunc").field(aggFuncInfo._2)
    case "sum" => aggregation.sum("aggFunc").field(aggFuncInfo._2)
    case "cardinality" => aggregation.cardinality("aggFunc").field(aggFuncInfo._2)
    case _ => aggregation.count("aggFunc").field(aggFuncInfo._2)
  }
}

//For parsing aggregation
def parseAgg[T](groups: Seq[String], agg: Aggregation, allGroups: Seq[String])(leafAggParser: (Seq[String], Aggregation) => Seq[T]): Seq[T] = {
  groups match {
    case x :: xs => {
      val groupAggs = agg.asInstanceOf[StringTerms].getBuckets.asScala.toList
      (for {
        groupAgg <- groupAggs
        aa = groupAgg.getAggregations.asList.asScala.head
        gkey = groupAgg.getKeyAsString
        gacc = allGroups :+ gkey
      } yield parseAgg(xs, aa, gacc)(leafAggParser)).flatten
    }

    case Nil => {
      leafAggParser(allGroups, agg)
    }
  }
}

//An example leaf aggregation parser
def parseSimpleLeafAgg(allGroups: Seq[String], agg: Aggregation): Seq[GroupStats] = {
  val value = agg.asInstanceOf[InternalNumericMetricsAggregation.SingleValue].value()
  val groupId = allGroups.mkString(".")
  Seq(GroupStats(groupId, value))
}

//Usage: Build Query and Parse result
def groupStats(groupFields: Seq[String]): Seq[GroupStats] = {
  val resp = client.execute {
    def leafPlainAggBuilder = buildLeafAgg(("count", "V1")) _
    search(esIndex).size(0).aggregations(buildAgg(groupFields)(leafPlainAggBuilder))
  }.await
  //get the root aggregation
  val agg = resp.aggregations.asList().asScala.head
  def leafAggParser = parseSimpleLeafAgg _
  val res = parseAgg(groupFields, agg, Seq())(leafAggParser)
  res
}