Extend WriteMetricsTransform to write feature value stats to StatsD

Author: davidheryanto

.prow/scripts/test-end-to-end.sh

@@ -67,24 +67,24 @@ tail -n10 /var/log/kafka.log
 kafkacat -b localhost:9092 -L
 
 if [[ ${SKIP_BUILD_JARS} != "true" ]]; then
-  echo "
-  ============================================================
-  Building jars for Feast
-  ============================================================
-  "
-
-  .prow/scripts/download-maven-cache.sh \
-      --archive-uri gs://feast-templocation-kf-feast/.m2.2019-10-24.tar \
-      --output-dir /root/
-
-  # Build jars for Feast
-  mvn --quiet --batch-mode --define skipTests=true clean package
-
-  ls -lh core/target/*jar
-  ls -lh serving/target/*jar
-  else
-    echo "[DEBUG] Skipping building jars"
-  fi
+echo "
+============================================================
+Building jars for Feast
+============================================================
+"
+
+.prow/scripts/download-maven-cache.sh \
+    --archive-uri gs://feast-templocation-kf-feast/.m2.2019-10-24.tar \
+    --output-dir /root/
+
+# Build jars for Feast
+mvn --quiet --batch-mode --define skipTests=true clean package
+
+ls -lh core/target/*jar
+ls -lh serving/target/*jar
+else
+  echo "[DEBUG] Skipping building jars"
+fi
 
 echo "
 ============================================================

ingestion/pom.xml

@@ -248,5 +248,12 @@
       <version>2.8.1</version>
     </dependency>
 
+    <!-- For calculation of percentiles in feature values -->
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-math3</artifactId>
+      <version>3.6.1</version>
+    </dependency>
+
   </dependencies>
 </project>

ingestion/src/main/java/feast/ingestion/options/ImportOptions.java

@@ -24,8 +24,11 @@
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.Validation.Required;
 
-/** Options passed to Beam to influence the job's execution environment */
+/**
+ * Options passed to Beam to influence the job's execution environment
+ */
 public interface ImportOptions extends PipelineOptions, DataflowPipelineOptions, DirectOptions {
+
   @Required
   @Description(
       "JSON string representation of the FeatureSet that the import job will process, in BZip2 binary format."
@@ -60,7 +63,7 @@ public interface ImportOptions extends PipelineOptions, DataflowPipelineOptions,
 
   /**
    * @param deadLetterTableSpec (Optional) BigQuery table for storing elements that failed to be
-   *     processed. Table spec must follow this format PROJECT_ID:DATASET_ID.PROJECT_ID
+   *                            processed. Table spec must follow this format PROJECT_ID:DATASET_ID.PROJECT_ID
    */
   void setDeadLetterTableSpec(String deadLetterTableSpec);
 
@@ -83,4 +86,13 @@ public interface ImportOptions extends PipelineOptions, DataflowPipelineOptions,
   int getStatsdPort();
 
   void setStatsdPort(int StatsdPort);
+
+  @Description(
+      "Fixed window size in seconds (default 30) to apply before aggregation of numerical value of features"
+          + "and writing the aggregated value to StatsD. Refer to feast.ingestion.transform.metrics.WriteFeatureValueMetricsDoFn"
+          + "for details on the metric names and types.")
+  @Default.Integer(30)
+  int getWindowSizeInSecForFeatureValueMetric();
+
+  void setWindowSizeInSecForFeatureValueMetric(int seconds);
 }