Added aggregate & conditonal readers for parquet

readers/src/main/scala/com/salesforce/op/readers/DataReaders.scala

@@ -170,6 +170,26 @@ object DataReaders {
       aggregateParams: AggregateParams[T]
     ): AggregateCSVProductReader[T] = csvProduct(path, key, aggregateParams)
 
+    /**
+     * Creates a [[AggregateParquetProductReader]]
+     */
+    def parquetProduct[T <: Product : Encoder : WeakTypeTag](
+      path: Option[String] = None,
+      key: T => String = randomKey _,
+      aggregateParams: AggregateParams[T]
+    ): AggregateParquetProductReader[T] = new AggregateParquetProductReader[T](
+      readPath = path, key = key, aggregateParams = aggregateParams
+    )
+
+    /**
+     * Creates a [[AggregateParquetProductReader]], but is called parquetCase so it's easier to understand
+     */
+    def parquetCase[T <: Product : Encoder : WeakTypeTag](
+      path: Option[String] = None,
+      key: T => String = randomKey _,
+      aggregateParams: AggregateParams[T]
+   ): AggregateParquetProductReader[T] = parquetProduct(path, key, aggregateParams)
+
   }
 
   /**
@@ -232,6 +252,27 @@ object DataReaders {
       conditionalParams: ConditionalParams[T]
     ): ConditionalCSVProductReader[T] = csvProduct(path, key, conditionalParams)
 
+    /**
+     * Creates a [[ConditionalParquetProductReader]]
+     */
+    def parquetProduct[T <: Product : Encoder : WeakTypeTag]
+    (
+      path: Option[String] = None,
+      key: T => String = randomKey _,
+      conditionalParams: ConditionalParams[T]
+    ): ConditionalParquetProductReader[T] = new ConditionalParquetProductReader[T](
+      readPath = path, key = key, conditionalParams = conditionalParams
+    )
+
+    /**
+     * Creates a [[ConditionalParquetProductReader]], but is called parquetCase so is easier to understand
+     */
+    def parquetCase[T <: Product : Encoder : WeakTypeTag](
+      path: Option[String] = None,
+      key: T => String = randomKey _,
+      conditionalParams: ConditionalParams[T]
+   ): ConditionalParquetProductReader[T] = parquetProduct(path, key, conditionalParams)
+
   }
 
 }

readers/src/main/scala/com/salesforce/op/readers/ParquetProductReader.scala

@@ -56,3 +56,33 @@ class ParquetProductReader[T <: Product : Encoder]
     maybeRepartition(data, params)
   }
 }
+
+/**
+  * Data Reader for Parquet events, where there may be multiple records for a given key. Each parquet record
+  * will be automatically converted to type T that defines an [[Encoder]].
+  * @param readPath default path to data
+  * @param key      function for extracting key from record
+  * @param aggregateParams params for time-based aggregation
+  * @tparam T
+  */
+class AggregateParquetProductReader[T <: Product : Encoder : WeakTypeTag]
+(
+  readPath: Option[String],
+  key: T => String,
+  val aggregateParams: AggregateParams[T]
+)extends ParquetProductReader[T](readPath, key) with AggregateDataReader[T]
+
+/**
+  * Data Reader for Parquet events, when computing conditional probabilities. There may be multiple records for
+  * a given key. Each parquet record will be automatically converted to type T that defines an [[Encoder]].
+  * @param readPath default path to data
+  * @param key      function for extracting key from record
+  * @param conditionalParams params for conditional aggregation
+  * @tparam T
+  */
+class ConditionalParquetProductReader[T <: Product : Encoder : WeakTypeTag]
+(
+  readPath: Option[String],
+  key: T => String,
+  val conditionalParams: ConditionalParams[T]
+)extends ParquetProductReader[T](readPath, key) with ConditionalDataReader[T]