Skip to content

Commit 4a9d751

Browse files
timstims
committed
Fix error handling and do not use empty string for tuple tags
1 parent 39d76a8 commit 4a9d751

10 files changed

Lines changed: 221 additions & 131 deletions

File tree

‎ingestion/src/main/java/feast/ingestion/ImportJob.java‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
package feast.ingestion;
1919

2020
import com.google.api.services.bigquery.model.TableRow;
21+
import com.google.common.collect.Lists;
2122
import com.google.inject.Guice;
2223
import com.google.inject.Inject;
2324
import com.google.inject.Injector;
@@ -42,6 +43,7 @@
4243
import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
4344
import feast.types.FeatureRowProto.FeatureRow;
4445
import java.util.Arrays;
46+
import java.util.List;
4547
import java.util.Random;
4648
import lombok.extern.slf4j.Slf4j;
4749
import org.apache.beam.runners.dataflow.DataflowPipelineJob;
@@ -53,13 +55,15 @@
5355
import org.apache.beam.sdk.extensions.protobuf.ProtoCoder;
5456
import org.apache.beam.sdk.io.gcp.bigquery.TableRowJsonCoder;
5557
import org.apache.beam.sdk.options.PipelineOptionsFactory;
58+
import org.apache.beam.sdk.transforms.Flatten;
5659
import org.apache.beam.sdk.transforms.ParDo;
5760
import org.apache.beam.sdk.transforms.Sample;
5861
import org.apache.beam.sdk.transforms.windowing.AfterWatermark;
5962
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
6063
import org.apache.beam.sdk.transforms.windowing.Window;
6164
import org.apache.beam.sdk.values.PCollection;
6265
import org.apache.beam.sdk.values.PCollection.IsBounded;
66+
import org.apache.beam.sdk.values.PCollectionList;
6367
import org.apache.beam.sdk.values.TypeDescriptor;
6468
import org.apache.commons.codec.digest.DigestUtils;
6569
import org.joda.time.DateTime;
@@ -165,17 +169,23 @@ public void expand() {
165169
"Round event timestamps to granularity",
166170
ParDo.of(new RoundEventTimestampsDoFn())),
167171
pFeatureRows.getErrors());
172+
168173
if (!dryRun) {
174+
List<PCollection<FeatureRowExtended>> errors = Lists.newArrayList();
169175
pFeatureRows = pFeatureRows.apply("Write to Serving Stores", servingStoreTransform);
170-
pFeatureRows.getErrors().apply("Write serving errors", errorsStoreTransform);
176+
errors.add(pFeatureRows.getErrors());
177+
pFeatureRows = PFeatureRows.of(pFeatureRows.getMain());
171178

172179
log.info(
173180
"A sample of any 2 rows from each of MAIN, RETRIES and ERRORS will logged for convenience");
174181
logNRows(pFeatureRows, "Output sample", 2);
175182

176183
PFeatureRows.of(pFeatureRows.getMain())
177184
.apply("Write to Warehouse Stores", warehouseStoreTransform);
178-
pFeatureRows.getErrors().apply("Write warehouse errors", errorsStoreTransform);
185+
errors.add(pFeatureRows.getErrors());
186+
187+
PCollectionList.of(errors).apply("flatten errors", Flatten.pCollections())
188+
.apply("Write serving errors", errorsStoreTransform);
179189
}
180190
}
181191

‎ingestion/src/main/java/feast/ingestion/transform/ErrorsStoreTransform.java‎

Lines changed: 26 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -17,72 +17,66 @@
1717

1818
package feast.ingestion.transform;
1919

20+
import static com.google.common.base.Preconditions.checkArgument;
2021
import static feast.ingestion.util.JsonUtil.convertJsonStringToMap;
2122

2223
import com.google.inject.Inject;
2324
import feast.ingestion.model.Specs;
2425
import feast.ingestion.options.ImportJobOptions;
2526
import feast.ingestion.transform.FeatureIO.Write;
26-
import feast.ingestion.transform.fn.LoggerDoFn;
2727
import feast.specs.StorageSpecProto.StorageSpec;
2828
import feast.storage.ErrorsStore;
2929
import feast.storage.noop.NoOpIO;
3030
import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
3131
import java.util.List;
3232
import lombok.extern.slf4j.Slf4j;
33-
import org.apache.beam.sdk.transforms.ParDo;
3433
import org.apache.beam.sdk.values.PCollection;
3534
import org.apache.beam.sdk.values.PDone;
36-
import org.slf4j.event.Level;
35+
import org.apache.hadoop.hbase.util.Strings;
3736

3837
@Slf4j
3938
public class ErrorsStoreTransform extends FeatureIO.Write {
4039

41-
public static final String ERRORS_STORE_STDERR = "stderr";
42-
public static final String ERRORS_STORE_STDOUT = "stdout";
43-
public static final String ERRORS_STORE_JSON = "file.json";
44-
4540
private String errorsStoreType;
4641
private StorageSpec errorsStoreSpec;
47-
private ErrorsStore errorsStore;
4842
private Specs specs;
43+
private List<ErrorsStore> errorsStores;
4944

5045
@Inject
5146
public ErrorsStoreTransform(
5247
ImportJobOptions options, Specs specs, List<ErrorsStore> errorsStores) {
5348
this.specs = specs;
49+
this.errorsStores = errorsStores;
5450
this.errorsStoreType = options.getErrorsStoreType();
5551

56-
for (ErrorsStore errorsStore : errorsStores) {
57-
if (errorsStore.getType().equals(errorsStoreType)) {
58-
this.errorsStore = errorsStore;
59-
}
52+
if (!Strings.isEmpty(errorsStoreType)) {
53+
this.errorsStoreSpec =
54+
StorageSpec.newBuilder()
55+
.setType(errorsStoreType)
56+
.putAllOptions(convertJsonStringToMap(options.getErrorsStoreOptions()))
57+
.build();
6058
}
61-
62-
this.errorsStoreSpec =
63-
StorageSpec.newBuilder()
64-
.setType(errorsStoreType)
65-
.putAllOptions(convertJsonStringToMap(options.getErrorsStoreOptions()))
66-
.build();
6759
}
6860

6961
@Override
7062
public PDone expand(PCollection<FeatureRowExtended> input) {
71-
switch (errorsStoreType) {
72-
case ERRORS_STORE_STDOUT:
73-
input.apply("Log errors to STDOUT", ParDo.of(new LoggerDoFn(Level.INFO)));
74-
break;
75-
case ERRORS_STORE_STDERR:
76-
input.apply("Log errors to STDERR", ParDo.of(new LoggerDoFn(Level.ERROR)));
77-
break;
78-
default:
79-
if (errorsStore == null) {
80-
log.warn("No valid errors store specified, errors will be discarded");
81-
return input.apply(new NoOpIO.Write());
82-
}
83-
Write write = errorsStore.create(this.errorsStoreSpec, specs);
84-
return input.apply(write);
63+
Write write;
64+
if (Strings.isEmpty(errorsStoreType)) {
65+
write = new NoOpIO.Write();
66+
} else {
67+
write = getErrorStore().create(this.errorsStoreSpec, specs);
8568
}
69+
input.apply("errors to " + String.valueOf(errorsStoreType), write);
8670
return PDone.in(input.getPipeline());
8771
}
72+
73+
ErrorsStore getErrorStore() {
74+
checkArgument(!errorsStoreType.isEmpty(), "Errors store type not provided");
75+
for (ErrorsStore errorsStore : errorsStores) {
76+
if (errorsStore.getType().equals(errorsStoreType)) {
77+
return errorsStore;
78+
}
79+
}
80+
throw new IllegalArgumentException("Errors store type not found");
81+
}
8882
}

‎ingestion/src/main/java/feast/ingestion/transform/SplitOutputByStore.java‎

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,6 @@
4141
import org.apache.beam.sdk.transforms.DoFn;
4242
import org.apache.beam.sdk.transforms.Flatten;
4343
import org.apache.beam.sdk.transforms.PTransform;
44-
import org.apache.beam.sdk.transforms.ParDo;
4544
import org.apache.beam.sdk.transforms.SerializableFunction;
4645
import org.apache.beam.sdk.values.PCollection;
4746
import org.apache.beam.sdk.values.PCollectionList;
@@ -59,7 +58,6 @@ public class SplitOutputByStore extends PTransform<PFeatureRows, PFeatureRows> {
5958
@Override
6059
public PFeatureRows expand(PFeatureRows input) {
6160
Map<String, Write> transforms = getFeatureStoreTransforms();
62-
transforms.put("", new NoOpIO.Write());
6361
Set<String> keys = transforms.keySet();
6462
Preconditions.checkArgument(transforms.size() > 0, "no write transforms found");
6563

@@ -72,6 +70,7 @@ public PFeatureRows expand(PFeatureRows input) {
7270
TupleTag<FeatureRowExtended> tag = splitter.getStrategy().getTag(key);
7371
taggedTransforms.put(tag, transforms.get(key));
7472
}
73+
7574
PFeatureRows output = splits.apply(new WriteTags(taggedTransforms, MultiOutputSplit.MAIN_TAG));
7675
return new PFeatureRows(
7776
output.getMain(),
@@ -122,13 +121,13 @@ public PFeatureRows expand(PCollectionTuple input) {
122121
mainList.add(main);
123122
}
124123

125-
String message =
126-
"FeatureRows have no matching write transform, these rows should not have passed validation.";
127-
PCollection<FeatureRowExtended> errors =
128-
input.get(mainTag).apply(ParDo.of(new WithErrors(getName(), message)));
129-
130-
return new PFeatureRows(
131-
PCollectionList.of(mainList).apply("Flatten main", Flatten.pCollections()), errors);
124+
/*
125+
* FeatureRows with no matching write transform `input.get(mainTag)` are considered
126+
* discardible, if they didn't have a matching store, they should have been discarded before
127+
* reaching here. So these will be feature with no output store at all.
128+
*/
129+
return PFeatureRows.of(
130+
PCollectionList.of(mainList).apply("Flatten main", Flatten.pCollections()));
132131
}
133132
}
134133

‎ingestion/src/main/java/feast/ingestion/values/PFeatureRows.java‎

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,11 @@
1717

1818
package feast.ingestion.values;
1919

20+
import feast.ingestion.transform.fn.BaseFeatureDoFn;
21+
import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
2022
import java.util.HashMap;
2123
import java.util.Map;
24+
import java.util.concurrent.atomic.AtomicInteger;
2225
import lombok.AllArgsConstructor;
2326
import lombok.Value;
2427
import lombok.extern.slf4j.Slf4j;
@@ -37,8 +40,6 @@
3740
import org.apache.beam.sdk.values.PValue;
3841
import org.apache.beam.sdk.values.TupleTag;
3942
import org.apache.beam.sdk.values.TupleTagList;
40-
import feast.ingestion.transform.fn.BaseFeatureDoFn;
41-
import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
4243

4344
@AllArgsConstructor
4445
@Value
@@ -47,14 +48,15 @@ public class PFeatureRows implements PInput, POutput {
4748

4849
public static final TupleTag<FeatureRowExtended> MAIN_TAG = new TupleTag<>();
4950
public static final TupleTag<FeatureRowExtended> ERRORS_TAG = new TupleTag<>();
50-
51+
private static final AtomicInteger counter = new AtomicInteger();
5152
private PCollection<FeatureRowExtended> main;
5253
private PCollection<FeatureRowExtended> errors;
5354

5455
public static PFeatureRows of(PCollection<FeatureRowExtended> input) {
5556
Pipeline pipeline = input.getPipeline();
5657
Create.Values<FeatureRowExtended> empty = Create.empty(ProtoCoder.of(FeatureRowExtended.class));
57-
return new PFeatureRows(input, pipeline.apply(input.getName() + "/empty.errors", empty));
58+
return new PFeatureRows(input,
59+
pipeline.apply(input.getName() + "/empty.errors" + counter.incrementAndGet(), empty));
5860
}
5961

6062
public static PFeatureRows of(
@@ -77,11 +79,12 @@ public Map<TupleTag<?>, PValue> expand() {
7779

7880
@Override
7981
public void finishSpecifyingOutput(
80-
String transformName, PInput input, PTransform<?, ?> transform) {}
82+
String transformName, PInput input, PTransform<?, ?> transform) {
83+
}
8184

8285
/**
8386
* @return new PFeatureRows, which has any tagged errors and retries in DoFn added to the errors
84-
* and retries gathered so far.
87+
* and retries gathered so far.
8588
*/
8689
public PFeatureRows applyDoFn(String name, BaseFeatureDoFn doFn) {
8790
MultiOutput<FeatureRowExtended, FeatureRowExtended> transform =
@@ -95,7 +98,7 @@ public PFeatureRows applyDoFn(String name, BaseFeatureDoFn doFn) {
9598

9699
PCollection<FeatureRowExtended> outErrors =
97100
PCollectionList.of(
98-
transformed.get(ERRORS_TAG).setCoder(ProtoCoder.of(FeatureRowExtended.class)))
101+
transformed.get(ERRORS_TAG).setCoder(ProtoCoder.of(FeatureRowExtended.class)))
99102
.and(errors)
100103
.apply(name + "/Flatten errors", Flatten.pCollections())
101104
.setCoder(ProtoCoder.of(FeatureRowExtended.class));

‎ingestion/src/main/java/feast/storage/noop/NoOpIO.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,9 @@
1818
package feast.storage.noop;
1919

2020
import feast.ingestion.transform.FeatureIO;
21+
import feast.ingestion.transform.fn.Identity;
2122
import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
23+
import org.apache.beam.sdk.transforms.ParDo;
2224
import org.apache.beam.sdk.values.PCollection;
2325
import org.apache.beam.sdk.values.PDone;
2426

@@ -28,6 +30,7 @@ public static class Write extends FeatureIO.Write {
2830

2931
@Override
3032
public PDone expand(PCollection<FeatureRowExtended> input) {
33+
input.apply(getName(), ParDo.of(new Identity(getName())));
3134
return PDone.in(input.getPipeline());
3235
}
3336
}

‎ingestion/src/main/java/feast/storage/stderr/LogIO.java‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
/*
2+
* Copyright 2018 The Feast Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*
16+
*/
17+
18+
package feast.storage.stderr;
19+
20+
import feast.ingestion.transform.FeatureIO;
21+
import feast.ingestion.transform.fn.LoggerDoFn;
22+
import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
23+
import lombok.AllArgsConstructor;
24+
import org.apache.beam.sdk.transforms.ParDo;
25+
import org.apache.beam.sdk.values.PCollection;
26+
import org.apache.beam.sdk.values.PDone;
27+
import org.slf4j.event.Level;
28+
29+
public class LogIO {
30+
31+
@AllArgsConstructor
32+
public static class Write extends FeatureIO.Write {
33+
34+
private Level level;
35+
36+
@Override
37+
public PDone expand(PCollection<FeatureRowExtended> input) {
38+
input.apply("Log to " + level.toString(), ParDo.of(new LoggerDoFn(level)));
39+
return PDone.in(input.getPipeline());
40+
}
41+
}
42+
}

‎ingestion/src/main/java/feast/storage/stderr/StderrErrorsStore.java‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
/*
2+
* Copyright 2018 The Feast Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*
16+
*/
17+
18+
package feast.storage.stderr;
19+
20+
import com.google.auto.service.AutoService;
21+
import feast.ingestion.model.Specs;
22+
import feast.ingestion.transform.FeatureIO.Write;
23+
import feast.specs.StorageSpecProto.StorageSpec;
24+
import feast.storage.ErrorsStore;
25+
import org.slf4j.event.Level;
26+
27+
@AutoService(ErrorsStore.class)
28+
public class StderrErrorsStore implements ErrorsStore {
29+
30+
public static final String TYPE_STDERR = "stderr";
31+
32+
@Override
33+
public Write create(StorageSpec storageSpec, Specs specs) {
34+
return new LogIO.Write(Level.ERROR);
35+
}
36+
37+
@Override
38+
public String getType() {
39+
return TYPE_STDERR;
40+
}
41+
}

0 commit comments

Comments
 (0)