feast-dev
diff --git a/‎ingestion/src/main/java/feast/ingestion/ImportJob.java‎
Lines changed: 11 additions & 0 deletions b/‎ingestion/src/main/java/feast/ingestion/ImportJob.java‎
Lines changed: 11 additions & 0 deletions
diff --git a/‎ingestion/src/main/java/feast/ingestion/transform/FilterFeatureRow.java‎
Lines changed: 0 additions & 52 deletions b/‎ingestion/src/main/java/feast/ingestion/transform/FilterFeatureRow.java‎
Lines changed: 0 additions & 52 deletions
diff --git a/‎ingestion/src/main/java/feast/ingestion/transform/ReadFeatureRow.java‎
Lines changed: 0 additions & 132 deletions b/‎ingestion/src/main/java/feast/ingestion/transform/ReadFeatureRow.java‎
Lines changed: 0 additions & 132 deletions
diff --git a/‎ingestion/src/main/java/feast/ingestion/transform/ToFeatureRowExtended.java‎
Lines changed: 0 additions & 48 deletions b/‎ingestion/src/main/java/feast/ingestion/transform/ToFeatureRowExtended.java‎
Lines changed: 0 additions & 48 deletions
diff --git a/‎ingestion/src/main/java/feast/ingestion/transform/WriteFeaturesTransform.java‎
Lines changed: 0 additions & 85 deletions b/‎ingestion/src/main/java/feast/ingestion/transform/WriteFeaturesTransform.java‎
Lines changed: 0 additions & 85 deletions
diff --git a/‎ingestion/src/main/java/feast/ingestion/transform/metrics/WindowRecords.java‎
Lines changed: 35 additions & 0 deletions b/‎ingestion/src/main/java/feast/ingestion/transform/metrics/WindowRecords.java‎
Lines changed: 35 additions & 0 deletions
@@ -7,6 +7,7 @@
 import feast.ingestion.transform.ReadFromSource;
 import feast.ingestion.transform.WriteFailedElementToBigQuery;
 import feast.ingestion.transform.WriteToStore;
+import feast.ingestion.transform.metrics.WriteMetricsTransform;
 import feast.ingestion.utils.ResourceUtil;
 import feast.ingestion.utils.SpecUtil;
 import feast.ingestion.utils.StoreUtil;
@@ -47,6 +48,7 @@ public static PipelineResult runPipeline(ImportOptions options)
      * 1. Read messages from Feast Source as FeatureRow
      * 2. Write FeatureRow to the corresponding Store
      * 3. Write elements that failed to be processed to a dead letter queue.
+     * 4. Write metrics to a metrics sink
      */
 
     PipelineOptionsValidator.validate(ImportOptions.class, options);
@@ -95,6 +97,15 @@ public static PipelineResult runPipeline(ImportOptions options)
                       .setTableSpec(options.getDeadLetterTableSpec())
                       .build());
         }
+
+        // Step 4. Write metrics to a metrics sink.
+        convertedFeatureRows
+            .apply("WriteMetrics", WriteMetricsTransform.newBuilder()
+                .setFeatureSetSpec(featureSet)
+                .setStoreName(store.getName())
+                .setSuccessTag(FEATURE_ROW_OUT)
+                .setFailureTag(DEADLETTER_OUT)
+                .build());
       }
     }
 
 
@@ -0,0 +1,35 @@
+package feast.ingestion.transform.metrics;
+
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.Duration;
+
+public class WindowRecords<T> extends
+    PTransform<PCollection<T>, PCollection<KV<Integer, Iterable<T>>>> {
+
+  private final long windowSize;
+
+  public WindowRecords(long windowSize) {
+    this.windowSize = windowSize;
+  }
+
+  @Override
+  public PCollection<KV<Integer, Iterable<T>>> expand(PCollection<T> input) {
+    return input
+        .apply("Window records",
+            Window.into(FixedWindows.of(Duration.standardSeconds(windowSize))))
+        .apply("Add key", ParDo.of(new DoFn<T, KV<Integer, T>>() {
+          @ProcessElement
+          public void processElement(ProcessContext c) {
+            c.output(KV.of(1, c.element()));
+          }
+        }))
+        .apply("Collect", GroupByKey.create());
+  }
+}