Skip to content

Extend WriteMetricsTransform in Ingestion to write feature value stats to StatsD #486

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
feast-ci-bot merged 6 commits into from
Feb 25, 2020
Merged

Extend WriteMetricsTransform in Ingestion to write feature value stats to StatsD #486

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
a87eee2
Extend WriteMetricsTransform to write feature value stats to StatsD
davidheryanto Feb 23, 2020
de32292
Apply mvn spotless
davidheryanto Feb 24, 2020
ed268eb
Catch all exception not just StatsDClientException during init
davidheryanto Feb 24, 2020
028b5eb
Change log level due to invalid feature set ref to error (previously …
davidheryanto Feb 25, 2020
413e569
Merge branch 'master' into feast-ingestion-feature-value-metric
davidheryanto Feb 25, 2020
0bbdfeb
Apply maven spotless to metric transform codes
davidheryanto Feb 25, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Apply maven spotless to metric transform codes
  • Loading branch information
@davidheryanto
davidheryanto committed Feb 25, 2020
commit 0bbdfeb7311ebc7715fc357f5625e7016ed00f8d
1 change: 0 additions & 1 deletion ingestion/src/main/java/feast/ingestion/transform/metrics/WriteFeatureValueMetricsDoFn.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@
import com.google.auto.value.AutoValue;
import com.timgroup.statsd.NonBlockingStatsDClient;
import com.timgroup.statsd.StatsDClient;
import com.timgroup.statsd.StatsDClientException;
import feast.types.FeatureRowProto.FeatureRow;
import feast.types.FieldProto.Field;
import feast.types.ValueProto.Value;
Expand Down
43 changes: 26 additions & 17 deletions ingestion/src/main/java/feast/ingestion/transform/metrics/WriteMetricsTransform.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,29 +86,39 @@ public PDone expand(PCollectionTuple input) {

// 1. Apply a fixed window
// 2. Group feature row by feature set reference
// 3. Calculate min, max, mean, percentiles of numerical values of features in the window and
// 3. Calculate min, max, mean, percentiles of numerical values of features in the window
// and
// 4. Send the aggregate value to StatsD metric collector.
//
// NOTE: window is applied here so the metric collector will not be overwhelmed with
// metrics data. And for metric data, only statistic of the values are usually required
// vs the actual values.
input
.get(getSuccessTag())
.apply("FixedWindow", Window.into(FixedWindows
.of(Duration.standardSeconds(options.getWindowSizeInSecForFeatureValueMetric()))))
.apply("ConvertTo_FeatureSetRefToFeatureRow",
ParDo.of(new DoFn<FeatureRow, KV<String, FeatureRow>>() {
@ProcessElement
public void processElement(ProcessContext c, @Element FeatureRow featureRow) {
c.output(KV.of(featureRow.getFeatureSet(), featureRow));
}
}))
.apply(
"FixedWindow",
Window.into(
FixedWindows.of(
Duration.standardSeconds(
options.getWindowSizeInSecForFeatureValueMetric()))))
.apply(
"ConvertTo_FeatureSetRefToFeatureRow",
ParDo.of(
new DoFn<FeatureRow, KV<String, FeatureRow>>() {
@ProcessElement
public void processElement(ProcessContext c, @Element FeatureRow featureRow) {
c.output(KV.of(featureRow.getFeatureSet(), featureRow));
}
}))
.apply("GroupByFeatureSetRef", GroupByKey.create())
.apply("WriteFeatureValueMetrics", ParDo.of(WriteFeatureValueMetricsDoFn.newBuilder()
.setStatsdHost(options.getStatsdHost())
.setStatsdPort(options.getStatsdPort())
.setStoreName(getStoreName())
.build()));
.apply(
"WriteFeatureValueMetrics",
ParDo.of(
WriteFeatureValueMetricsDoFn.newBuilder()
.setStatsdHost(options.getStatsdHost())
.setStatsdPort(options.getStatsdPort())
.setStoreName(getStoreName())
.build()));

return PDone.in(input.getPipeline());
case "none":
Expand All @@ -120,8 +130,7 @@ public void processElement(ProcessContext c, @Element FeatureRow featureRow) {
ParDo.of(
new DoFn<FeatureRow, Void>() {
@ProcessElement
public void processElement(ProcessContext c) {
}
public void processElement(ProcessContext c) {}
}));
return PDone.in(input.getPipeline());
}
Expand Down
1 change: 0 additions & 1 deletion ingestion/src/main/java/feast/ingestion/transform/metrics/WriteRowMetricsDoFn.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,6 @@ public abstract static class Builder {
@Setup
public void setup() {
statsd = new NonBlockingStatsDClient(METRIC_PREFIX, getStatsdHost(), getStatsdPort());

}

@ProcessElement
Expand Down