Adding aggregators

airbnb · Dec 7, 2023 · 2d497eb · 2d497eb
1 parent 3b3f1c1
commit 2d497eb
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 4 deletions.
diff --git a/aggregator/src/main/scala/ai/chronon/aggregator/windowing/HopsAggregator.scala b/aggregator/src/main/scala/ai/chronon/aggregator/windowing/HopsAggregator.scala
@@ -16,6 +16,7 @@
 
 package ai.chronon.aggregator.windowing
 
+import org.slf4j.LoggerFactory
 import ai.chronon.aggregator.row.RowAggregator
 import ai.chronon.aggregator.windowing.HopsAggregator._
 import ai.chronon.api.Extensions.{AggregationOps, AggregationsOps, WindowOps, WindowUtils}
@@ -93,6 +94,7 @@ class HopsAggregator(minQueryTs: Long,
                      inputSchema: Seq[(String, DataType)],
                      resolution: Resolution)
     extends HopsAggregatorBase(aggregations, inputSchema, resolution) {
+  private val logger = LoggerFactory.getLogger(getClass)
 
   val leftBoundaries: Array[Option[Long]] = {
     // Nikhil is pretty confident we won't call this when aggregations is empty
@@ -135,7 +137,7 @@ class HopsAggregator(minQueryTs: Long,
       .zip(readableLeftBounds)
       .map { case (hop, left) => s"$hop->$left" }
       .mkString(", ")
-    println(s"""Left bounds: $readableHopsToBoundsMap 
+    logger.info(s"""Left bounds: $readableHopsToBoundsMap 
          |minQueryTs = ${TsUtils.toStr(minQueryTs)}""".stripMargin)
     result
   }

diff --git a/aggregator/src/main/scala/ai/chronon/aggregator/windowing/SawtoothOnlineAggregator.scala b/aggregator/src/main/scala/ai/chronon/aggregator/windowing/SawtoothOnlineAggregator.scala
@@ -16,6 +16,7 @@
 
 package ai.chronon.aggregator.windowing
 
+import org.slf4j.LoggerFactory
 import scala.collection.Seq
 import ai.chronon.api.Extensions.{AggregationPartOps, WindowOps}
 import ai.chronon.api._
@@ -31,6 +32,7 @@ class SawtoothOnlineAggregator(val batchEndTs: Long,
                                        inputSchema: Seq[(String, DataType)],
                                        resolution: Resolution,
                                        tailBufferMillis: Long) {
+  private val logger = LoggerFactory.getLogger(getClass)
 
   // logically, batch response is arranged like so
   // sum-90d =>  sum_ir_88d, [(sum_ir_1d, ts)] -> 1d is the hopSize for 90d
@@ -46,10 +48,10 @@ class SawtoothOnlineAggregator(val batchEndTs: Long,
 
   val batchTailTs: Array[Option[Long]] = tailTs(batchEndTs)
 
-  println(s"Batch End: ${TsUtils.toStr(batchEndTs)}")
-  println("Window Tails: ")
+  logger.info(s"Batch End: ${TsUtils.toStr(batchEndTs)}")
+  logger.info("Window Tails: ")
   for (i <- windowMappings.indices) {
-    println(s"  ${windowMappings(i).aggregationPart.outputColumnName} -> ${batchTailTs(i).map(TsUtils.toStr)}")
+    logger.info(s"  ${windowMappings(i).aggregationPart.outputColumnName} -> ${batchTailTs(i).map(TsUtils.toStr)}")
   }
 
   def update(batchIr: BatchIr, row: Row): BatchIr = update(batchEndTs, batchIr, row, batchTailTs)