move partition consolidator and add LocalAggregator API

src/main/scala/com/microsoft/ml/spark/io/http/PartitionConsolidator.scala → src/main/scala/com/microsoft/ml/spark/stages/PartitionConsolidator.scala

@@ -1,14 +1,16 @@
 // Copyright (C) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License. See LICENSE in project root for information.
 
-package com.microsoft.ml.spark.io.http
+package com.microsoft.ml.spark.stages
 
 import java.util.concurrent.LinkedBlockingQueue
+
 import com.microsoft.ml.spark.core.contracts.{HasInputCol, HasOutputCol}
+import com.microsoft.ml.spark.io.http.{HTTPParams, SharedSingleton}
 import com.microsoft.ml.spark.logging.BasicLogging
-import org.apache.spark.ml.{ComplexParamsWritable, Transformer}
 import org.apache.spark.ml.param._
 import org.apache.spark.ml.util.{DefaultParamsReadable, Identifiable}
+import org.apache.spark.ml.{ComplexParamsWritable, Transformer}
 import org.apache.spark.sql.catalyst.encoders.RowEncoder
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.{DataFrame, Dataset, Row}
@@ -134,3 +136,8 @@ class PartitionConsolidator(val uid: String)
 
   override def transformSchema(schema: StructType): StructType = schema
 }
+
+trait LocalAggregator[T] {
+  def prep(iter: Iterator[Row]): T
+  def merge(ts: Seq[T]): T
+}

src/test/scala/com/microsoft/ml/spark/flaky/PartitionConsolidatorSuite.scala

@@ -5,7 +5,7 @@ package com.microsoft.ml.spark.flaky
 
 import com.microsoft.ml.spark.core.test.base.{SparkSessionFactory, TestBase, TimeLimitedFlaky}
 import com.microsoft.ml.spark.core.test.fuzzing.{TestObject, TransformerFuzzing}
-import com.microsoft.ml.spark.io.http.PartitionConsolidator
+import com.microsoft.ml.spark.stages.PartitionConsolidator
 import org.apache.spark.ml.util.MLReadable
 import org.apache.spark.sql.catalyst.encoders.RowEncoder
 import org.apache.spark.sql.types.{DoubleType, StructType}