1 /***********************************************************************
2  * Copyright (c) 2013-2024 Commonwealth Computer Research, Inc.
3  * All rights reserved. This program and the accompanying materials
4  * are made available under the terms of the Apache License, Version 2.0
5  * which accompanies this distribution and is available at
6  * http://www.opensource.org/licenses/apache2.0.php.
7  ***********************************************************************/
8 
9 package org.locationtech.geomesa.utils.stats
10 import com.typesafe.scalalogging.LazyLogging
11 import org.geotools.api.feature.simple.{SimpleFeature, SimpleFeatureType}
12 import org.locationtech.geomesa.utils.clearspring.StreamSummary
13 
14 import scala.collection.immutable.ListMap
15 
16 /**
17   * TopK stat
18   *
19   * @param sft simple feature type
20   * @param property property name of the attribute to track
21   * @param summary stream summary object
22   * @tparam T attribute type binding
23   */
24 class TopK[T] private [stats] (
25     val sft: SimpleFeatureType,
26     val property: String,
27     private [stats] val summary: StreamSummary[T]
28   ) extends Stat with LazyLogging {
29 
30   import TopK.StreamCapacity
31 
32   def this(sft: SimpleFeatureType, property: String) = this(sft, property, StreamSummary[T](TopK.StreamCapacity))
33 
34   override type S = TopK[T]
35 
36   private val i = sft.indexOf(property)
37 
38   def topK(k: Int): Iterator[(T, Long)] = summary.topK(k)
39   def size: Int = summary.size
40 
41   override def observe(sf: SimpleFeature): Unit = {
42     val value = sf.getAttribute(i).asInstanceOf[T]
43     if (value != null) {
44       summary.offer(value)
45     }
46   }
47 
48   override def unobserve(sf: SimpleFeature): Unit = {
49     val value = sf.getAttribute(i).asInstanceOf[T]
50     if (value != null) {
51       summary.offer(value, -1)
52     }
53   }
54 
55   override def +(other: TopK[T]): TopK[T] = {
56     val merged = new TopK[T](sft, property)
57     merged += this
58     merged += other
59     merged
60   }
61 
62   override def +=(other: TopK[T]): Unit =
63     other.summary.topK(StreamCapacity).foreach { case (item, count) => summary.offer(item, count) }
64 
65   override def clear(): Unit = summary.clear()
66 
67   override def isEmpty: Boolean = summary.size == 0
68 
69   override def toJsonObject: Any = {
70     val maps = summary.topK(10).zipWithIndex.map { case ((item, count), rank) =>
71       (rank, ListMap( "value" -> item, "count" -> count))
72     }
73     ListMap(maps.toSeq:_*)
74   }
75 
76   override def isEquivalent(other: Stat): Boolean = other match {
77     case s: TopK[T] if summary.size == s.summary.size =>
78       s.summary.topK(summary.size).sameElements(summary.topK(summary.size))
79     case _ => false
80   }
81 }
82 
83 object TopK {
84   val StreamCapacity = 1000
85 }