Skip to content

Commit 08cdfb4

Browse files
davidheryantokhorshuheng
davidheryanto
authored and
khorshuheng
committed
Apply a fixed window before writing row metrics (#590)
1 parent 3c8b2fd commit 08cdfb4

8 files changed

Lines changed: 413 additions & 166 deletions

File tree

‎ingestion/src/main/java/feast/ingestion/transform/metrics/WriteFeatureValueMetricsDoFn.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,11 +119,14 @@ public void processElement(
119119
ProcessContext context,
120120
@Element KV<String, Iterable<FeatureRow>> featureSetRefToFeatureRows) {
121121
if (statsDClient == null) {
122+
log.error("StatsD client is null, likely because it encounters an error during setup");
122123
return;
123124
}
124125

125126
String featureSetRef = featureSetRefToFeatureRows.getKey();
126127
if (featureSetRef == null) {
128+
log.error(
129+
"Feature set reference in the feature row is null. Please check the input feature rows from previous steps");
127130
return;
128131
}
129132
String[] colonSplits = featureSetRef.split(":");

‎ingestion/src/main/java/feast/ingestion/transform/metrics/WriteMetricsTransform.java‎

Lines changed: 41 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
2828
import org.apache.beam.sdk.transforms.windowing.Window;
2929
import org.apache.beam.sdk.values.KV;
30+
import org.apache.beam.sdk.values.PCollection;
3031
import org.apache.beam.sdk.values.PCollectionTuple;
3132
import org.apache.beam.sdk.values.PDone;
3233
import org.apache.beam.sdk.values.TupleTag;
@@ -73,52 +74,47 @@ public PDone expand(PCollectionTuple input) {
7374
.setStoreName(getStoreName())
7475
.build()));
7576

76-
input
77-
.get(getSuccessTag())
78-
.apply(
79-
"WriteRowMetrics",
80-
ParDo.of(
81-
WriteRowMetricsDoFn.newBuilder()
82-
.setStatsdHost(options.getStatsdHost())
83-
.setStatsdPort(options.getStatsdPort())
84-
.setStoreName(getStoreName())
85-
.build()));
77+
// Fixed window is applied so the metric collector will not be overwhelmed with the metrics
78+
// data. For validation, only summaries of the values are usually required vs the actual
79+
// values.
80+
PCollection<KV<String, Iterable<FeatureRow>>> validRowsGroupedByRef =
81+
input
82+
.get(getSuccessTag())
83+
.apply(
84+
"FixedWindow",
85+
Window.into(
86+
FixedWindows.of(
87+
Duration.standardSeconds(
88+
options.getWindowSizeInSecForFeatureValueMetric()))))
89+
.apply(
90+
"ConvertToKV_FeatureSetRefToFeatureRow",
91+
ParDo.of(
92+
new DoFn<FeatureRow, KV<String, FeatureRow>>() {
93+
@ProcessElement
94+
public void processElement(
95+
ProcessContext c, @Element FeatureRow featureRow) {
96+
c.output(KV.of(featureRow.getFeatureSet(), featureRow));
97+
}
98+
}))
99+
.apply("GroupByFeatureSetRef", GroupByKey.create());
86100

87-
// 1. Apply a fixed window
88-
// 2. Group feature row by feature set reference
89-
// 3. Calculate min, max, mean, percentiles of numerical values of features in the window
90-
// and
91-
// 4. Send the aggregate value to StatsD metric collector.
92-
//
93-
// NOTE: window is applied here so the metric collector will not be overwhelmed with
94-
// metrics data. And for metric data, only statistic of the values are usually required
95-
// vs the actual values.
96-
input
97-
.get(getSuccessTag())
98-
.apply(
99-
"FixedWindow",
100-
Window.into(
101-
FixedWindows.of(
102-
Duration.standardSeconds(
103-
options.getWindowSizeInSecForFeatureValueMetric()))))
104-
.apply(
105-
"ConvertTo_FeatureSetRefToFeatureRow",
106-
ParDo.of(
107-
new DoFn<FeatureRow, KV<String, FeatureRow>>() {
108-
@ProcessElement
109-
public void processElement(ProcessContext c, @Element FeatureRow featureRow) {
110-
c.output(KV.of(featureRow.getFeatureSet(), featureRow));
111-
}
112-
}))
113-
.apply("GroupByFeatureSetRef", GroupByKey.create())
114-
.apply(
115-
"WriteFeatureValueMetrics",
116-
ParDo.of(
117-
WriteFeatureValueMetricsDoFn.newBuilder()
118-
.setStatsdHost(options.getStatsdHost())
119-
.setStatsdPort(options.getStatsdPort())
120-
.setStoreName(getStoreName())
121-
.build()));
101+
validRowsGroupedByRef.apply(
102+
"WriteRowMetrics",
103+
ParDo.of(
104+
WriteRowMetricsDoFn.newBuilder()
105+
.setStatsdHost(options.getStatsdHost())
106+
.setStatsdPort(options.getStatsdPort())
107+
.setStoreName(getStoreName())
108+
.build()));
109+
110+
validRowsGroupedByRef.apply(
111+
"WriteFeatureValueMetrics",
112+
ParDo.of(
113+
WriteFeatureValueMetricsDoFn.newBuilder()
114+
.setStatsdHost(options.getStatsdHost())
115+
.setStatsdPort(options.getStatsdPort())
116+
.setStoreName(getStoreName())
117+
.build()));
122118

123119
return PDone.in(input.getPipeline());
124120
case "none":

‎ingestion/src/main/java/feast/ingestion/transform/metrics/WriteRowMetricsDoFn.java‎

Lines changed: 164 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -17,17 +17,26 @@
1717
package feast.ingestion.transform.metrics;
1818

1919
import com.google.auto.value.AutoValue;
20+
import com.google.protobuf.util.Timestamps;
2021
import com.timgroup.statsd.NonBlockingStatsDClient;
2122
import com.timgroup.statsd.StatsDClient;
22-
import com.timgroup.statsd.StatsDClientException;
2323
import feast.types.FeatureRowProto.FeatureRow;
2424
import feast.types.FieldProto.Field;
25+
import feast.types.ValueProto.Value;
2526
import feast.types.ValueProto.Value.ValCase;
27+
import java.time.Clock;
28+
import java.util.HashMap;
29+
import java.util.Map;
30+
import java.util.Map.Entry;
31+
import javax.annotation.Nullable;
2632
import org.apache.beam.sdk.transforms.DoFn;
33+
import org.apache.beam.sdk.values.KV;
34+
import org.apache.commons.lang3.ArrayUtils;
35+
import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
2736
import org.slf4j.Logger;
2837

2938
@AutoValue
30-
public abstract class WriteRowMetricsDoFn extends DoFn<FeatureRow, Void> {
39+
public abstract class WriteRowMetricsDoFn extends DoFn<KV<String, Iterable<FeatureRow>>, Void> {
3140

3241
private static final Logger log = org.slf4j.LoggerFactory.getLogger(WriteRowMetricsDoFn.class);
3342

@@ -39,12 +48,38 @@ public abstract class WriteRowMetricsDoFn extends DoFn<FeatureRow, Void> {
3948
public static final String FEATURE_TAG_KEY = "feast_feature_name";
4049
public static final String INGESTION_JOB_NAME_KEY = "ingestion_job_name";
4150

51+
public static final String GAUGE_NAME_FEATURE_ROW_LAG_MS_MIN = "feature_row_lag_ms_min";
52+
public static final String GAUGE_NAME_FEATURE_ROW_LAG_MS_MAX = "feature_row_lag_ms_max";
53+
public static final String GAUGE_NAME_FEATURE_ROW_LAG_MS_MEAN = "feature_row_lag_ms_mean";
54+
public static final String GAUGE_NAME_FEATURE_ROW_LAG_MS_PERCENTILE_90 =
55+
"feature_row_lag_ms_percentile_90";
56+
public static final String GAUGE_NAME_FEATURE_ROW_LAG_MS_PERCENTILE_95 =
57+
"feature_row_lag_ms_percentile_95";
58+
public static final String GAUGE_NAME_FEATURE_ROW_LAG_MS_PERCENTILE_99 =
59+
"feature_row_lag_ms_percentile_99";
60+
61+
public static final String GAUGE_NAME_FEATURE_VALUE_LAG_MS_MIN = "feature_value_lag_ms_min";
62+
public static final String GAUGE_NAME_FEATURE_VALUE_LAG_MS_MAX = "feature_value_lag_ms_max";
63+
public static final String GAUGE_NAME_FEATURE_VALUE_LAG_MS_MEAN = "feature_value_lag_ms_mean";
64+
public static final String GAUGE_NAME_FEATURE_VALUE_LAG_MS_PERCENTILE_90 =
65+
"feature_value_lag_ms_percentile_90";
66+
public static final String GAUGE_NAME_FEATURE_VALUE_LAG_MS_PERCENTILE_95 =
67+
"feature_value_lag_ms_percentile_95";
68+
public static final String GAUGE_NAME_FEATURE_VALUE_LAG_MS_PERCENTILE_99 =
69+
"feature_value_lag_ms_percentile_99";
70+
71+
public static final String COUNT_NAME_FEATURE_ROW_INGESTED = "feature_row_ingested_count";
72+
public static final String COUNT_NAME_FEATURE_VALUE_MISSING = "feature_value_missing_count";
73+
4274
public abstract String getStoreName();
4375

4476
public abstract String getStatsdHost();
4577

4678
public abstract int getStatsdPort();
4779

80+
@Nullable
81+
public abstract Clock getClock();
82+
4883
public static WriteRowMetricsDoFn create(
4984
String newStoreName, String newStatsdHost, int newStatsdPort) {
5085
return newBuilder()
@@ -69,79 +104,147 @@ public abstract static class Builder {
69104

70105
public abstract Builder setStatsdPort(int statsdPort);
71106

107+
/**
108+
* setClock will override the default system clock used to calculate feature row lag.
109+
*
110+
* @param clock Clock instance
111+
*/
112+
public abstract Builder setClock(Clock clock);
113+
72114
public abstract WriteRowMetricsDoFn build();
73115
}
74116

75117
@Setup
76118
public void setup() {
77-
statsd = new NonBlockingStatsDClient(METRIC_PREFIX, getStatsdHost(), getStatsdPort());
119+
// Note that exception may be thrown during StatsD client instantiation but no exception
120+
// will be thrown when sending metrics (mimicking the UDP protocol behaviour).
121+
// https://jar-download.com/artifacts/com.datadoghq/java-dogstatsd-client/2.1.1/documentation
122+
// https://github.com/DataDog/java-dogstatsd-client#unix-domain-socket-support
123+
try {
124+
statsd = new NonBlockingStatsDClient(METRIC_PREFIX, getStatsdHost(), getStatsdPort());
125+
} catch (Exception e) {
126+
log.error("StatsD client cannot be started: " + e.getMessage());
127+
}
78128
}
79129

130+
@SuppressWarnings("DuplicatedCode")
80131
@ProcessElement
81-
public void processElement(ProcessContext c) {
132+
public void processElement(
133+
ProcessContext c, @Element KV<String, Iterable<FeatureRow>> featureSetRefToFeatureRows) {
134+
if (statsd == null) {
135+
log.error("StatsD client is null, likely because it encounters an error during setup");
136+
return;
137+
}
82138

83-
try {
84-
FeatureRow row = c.element();
85-
long eventTimestamp = com.google.protobuf.util.Timestamps.toMillis(row.getEventTimestamp());
86-
87-
String[] split = row.getFeatureSet().split(":");
88-
String featureSetProject = split[0].split("/")[0];
89-
String featureSetName = split[0].split("/")[1];
90-
String featureSetVersion = split[1];
91-
92-
statsd.histogram(
93-
"feature_row_lag_ms",
94-
System.currentTimeMillis() - eventTimestamp,
95-
STORE_TAG_KEY + ":" + getStoreName(),
96-
FEATURE_SET_PROJECT_TAG_KEY + ":" + featureSetProject,
97-
FEATURE_SET_NAME_TAG_KEY + ":" + featureSetName,
98-
FEATURE_SET_VERSION_TAG_KEY + ":" + featureSetVersion,
99-
INGESTION_JOB_NAME_KEY + ":" + c.getPipelineOptions().getJobName());
100-
101-
statsd.histogram(
102-
"feature_row_event_time_epoch_ms",
103-
eventTimestamp,
104-
STORE_TAG_KEY + ":" + getStoreName(),
105-
FEATURE_SET_PROJECT_TAG_KEY + ":" + featureSetProject,
106-
FEATURE_SET_NAME_TAG_KEY + ":" + featureSetName,
107-
FEATURE_SET_VERSION_TAG_KEY + ":" + featureSetVersion,
108-
INGESTION_JOB_NAME_KEY + ":" + c.getPipelineOptions().getJobName());
109-
110-
for (Field field : row.getFieldsList()) {
111-
if (!field.getValue().getValCase().equals(ValCase.VAL_NOT_SET)) {
112-
statsd.histogram(
113-
"feature_value_lag_ms",
114-
System.currentTimeMillis() - eventTimestamp,
115-
STORE_TAG_KEY + ":" + getStoreName(),
116-
FEATURE_SET_PROJECT_TAG_KEY + ":" + featureSetProject,
117-
FEATURE_SET_NAME_TAG_KEY + ":" + featureSetName,
118-
FEATURE_SET_VERSION_TAG_KEY + ":" + featureSetVersion,
119-
FEATURE_TAG_KEY + ":" + field.getName(),
120-
INGESTION_JOB_NAME_KEY + ":" + c.getPipelineOptions().getJobName());
139+
String featureSetRef = featureSetRefToFeatureRows.getKey();
140+
if (featureSetRef == null) {
141+
log.error(
142+
"Feature set reference in the feature row is null. Please check the input feature rows from previous steps");
143+
return;
144+
}
145+
String[] colonSplits = featureSetRef.split(":");
146+
if (colonSplits.length != 2) {
147+
log.error(
148+
"Skip writing feature row metrics because the feature set reference '{}' does not"
149+
+ "follow the required format <project>/<feature_set_name>:<version>",
150+
featureSetRef);
151+
return;
152+
}
153+
String[] slashSplits = colonSplits[0].split("/");
154+
if (slashSplits.length != 2) {
155+
log.error(
156+
"Skip writing feature row metrics because the feature set reference '{}' does not"
157+
+ "follow the required format <project>/<feature_set_name>:<version>",
158+
featureSetRef);
159+
return;
160+
}
161+
162+
String featureSetProject = slashSplits[0];
163+
String featureSetName = slashSplits[1];
164+
String featureSetVersion = colonSplits[1];
165+
166+
// featureRowLagStats is stats for feature row lag for feature set "featureSetName"
167+
DescriptiveStatistics featureRowLagStats = new DescriptiveStatistics();
168+
// featureNameToLagStats is stats for feature lag for all features in feature set
169+
// "featureSetName"
170+
Map<String, DescriptiveStatistics> featureNameToLagStats = new HashMap<>();
171+
// featureNameToMissingCount is count for "value_not_set" for all features in feature set
172+
// "featureSetName"
173+
Map<String, Long> featureNameToMissingCount = new HashMap<>();
174+
175+
for (FeatureRow featureRow : featureSetRefToFeatureRows.getValue()) {
176+
long currentTime = getClock() == null ? System.currentTimeMillis() : getClock().millis();
177+
long featureRowLag = currentTime - Timestamps.toMillis(featureRow.getEventTimestamp());
178+
featureRowLagStats.addValue(featureRowLag);
179+
180+
for (Field field : featureRow.getFieldsList()) {
181+
String featureName = field.getName();
182+
Value featureValue = field.getValue();
183+
if (!featureNameToLagStats.containsKey(featureName)) {
184+
// Ensure map contains the "featureName" key
185+
featureNameToLagStats.put(featureName, new DescriptiveStatistics());
186+
}
187+
if (!featureNameToMissingCount.containsKey(featureName)) {
188+
// Ensure map contains the "featureName" key
189+
featureNameToMissingCount.put(featureName, 0L);
190+
}
191+
if (featureValue.getValCase().equals(ValCase.VAL_NOT_SET)) {
192+
featureNameToMissingCount.put(
193+
featureName, featureNameToMissingCount.get(featureName) + 1);
121194
} else {
122-
statsd.count(
123-
"feature_value_missing_count",
124-
1,
125-
STORE_TAG_KEY + ":" + getStoreName(),
126-
FEATURE_SET_PROJECT_TAG_KEY + ":" + featureSetProject,
127-
FEATURE_SET_NAME_TAG_KEY + ":" + featureSetName,
128-
FEATURE_SET_VERSION_TAG_KEY + ":" + featureSetVersion,
129-
FEATURE_TAG_KEY + ":" + field.getName(),
130-
INGESTION_JOB_NAME_KEY + ":" + c.getPipelineOptions().getJobName());
195+
featureNameToLagStats.get(featureName).addValue(featureRowLag);
131196
}
132197
}
198+
}
199+
200+
String[] tags = {
201+
STORE_TAG_KEY + ":" + getStoreName(),
202+
FEATURE_SET_PROJECT_TAG_KEY + ":" + featureSetProject,
203+
FEATURE_SET_NAME_TAG_KEY + ":" + featureSetName,
204+
FEATURE_SET_VERSION_TAG_KEY + ":" + featureSetVersion,
205+
INGESTION_JOB_NAME_KEY + ":" + c.getPipelineOptions().getJobName(),
206+
};
207+
208+
statsd.count(COUNT_NAME_FEATURE_ROW_INGESTED, featureRowLagStats.getN(), tags);
209+
// DescriptiveStatistics returns invalid NaN value for getMin(), getMax(), ... when there is no
210+
// items in the stats.
211+
if (featureRowLagStats.getN() > 0) {
212+
statsd.gauge(GAUGE_NAME_FEATURE_ROW_LAG_MS_MIN, featureRowLagStats.getMin(), tags);
213+
statsd.gauge(GAUGE_NAME_FEATURE_ROW_LAG_MS_MAX, featureRowLagStats.getMax(), tags);
214+
statsd.gauge(GAUGE_NAME_FEATURE_ROW_LAG_MS_MEAN, featureRowLagStats.getMean(), tags);
215+
statsd.gauge(
216+
GAUGE_NAME_FEATURE_ROW_LAG_MS_PERCENTILE_90, featureRowLagStats.getPercentile(90), tags);
217+
statsd.gauge(
218+
GAUGE_NAME_FEATURE_ROW_LAG_MS_PERCENTILE_95, featureRowLagStats.getPercentile(95), tags);
219+
statsd.gauge(
220+
GAUGE_NAME_FEATURE_ROW_LAG_MS_PERCENTILE_99, featureRowLagStats.getPercentile(99), tags);
221+
}
133222

223+
for (Entry<String, DescriptiveStatistics> entry : featureNameToLagStats.entrySet()) {
224+
String featureName = entry.getKey();
225+
String[] tagsWithFeatureName = ArrayUtils.add(tags, FEATURE_TAG_KEY + ":" + featureName);
226+
DescriptiveStatistics stats = entry.getValue();
227+
if (stats.getN() > 0) {
228+
statsd.gauge(GAUGE_NAME_FEATURE_VALUE_LAG_MS_MIN, stats.getMin(), tagsWithFeatureName);
229+
statsd.gauge(GAUGE_NAME_FEATURE_VALUE_LAG_MS_MAX, stats.getMax(), tagsWithFeatureName);
230+
statsd.gauge(GAUGE_NAME_FEATURE_VALUE_LAG_MS_MEAN, stats.getMean(), tagsWithFeatureName);
231+
statsd.gauge(
232+
GAUGE_NAME_FEATURE_VALUE_LAG_MS_PERCENTILE_90,
233+
stats.getPercentile(90),
234+
tagsWithFeatureName);
235+
statsd.gauge(
236+
GAUGE_NAME_FEATURE_VALUE_LAG_MS_PERCENTILE_95,
237+
stats.getPercentile(95),
238+
tagsWithFeatureName);
239+
statsd.gauge(
240+
GAUGE_NAME_FEATURE_VALUE_LAG_MS_PERCENTILE_99,
241+
stats.getPercentile(99),
242+
tagsWithFeatureName);
243+
}
134244
statsd.count(
135-
"feature_row_ingested_count",
136-
1,
137-
STORE_TAG_KEY + ":" + getStoreName(),
138-
FEATURE_SET_PROJECT_TAG_KEY + ":" + featureSetProject,
139-
FEATURE_SET_NAME_TAG_KEY + ":" + featureSetName,
140-
FEATURE_SET_VERSION_TAG_KEY + ":" + featureSetVersion,
141-
INGESTION_JOB_NAME_KEY + ":" + c.getPipelineOptions().getJobName());
142-
143-
} catch (StatsDClientException e) {
144-
log.warn("Unable to push metrics to server", e);
245+
COUNT_NAME_FEATURE_VALUE_MISSING,
246+
featureNameToMissingCount.get(featureName),
247+
tagsWithFeatureName);
145248
}
146249
}
147250
}

0 commit comments

Comments
 (0)