storageIds);
-
- interface Builder extends Serializable {
- SpecService build();
- }
-
- @AllArgsConstructor
- class UnsupportedBuilder implements Builder {
- private String message;
-
- @Override
- public SpecService build() {
- throw new UnsupportedOperationException(message);
- }
- }
-}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/CoalesceFeatureRows.java b/ingestion/src/main/java/feast/ingestion/transform/CoalesceFeatureRows.java
new file mode 100644
index 00000000000..aa8a1754c2f
--- /dev/null
+++ b/ingestion/src/main/java/feast/ingestion/transform/CoalesceFeatureRows.java
@@ -0,0 +1,293 @@
+/*
+ * Copyright 2019 The Feast Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package feast.ingestion.transform;
+
+import com.google.common.base.Preconditions;
+import com.google.protobuf.util.Timestamps;
+import feast.types.FeatureProto.Feature;
+import feast.types.FeatureRowProto.FeatureRow;
+import feast_ingestion.types.CoalesceAccumProto.CoalesceAccum;
+import feast_ingestion.types.CoalesceKeyProto.CoalesceKey;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.extensions.protobuf.ProtoCoder;
+import org.apache.beam.sdk.state.BagState;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.TimeDomain;
+import org.apache.beam.sdk.state.Timer;
+import org.apache.beam.sdk.state.TimerSpec;
+import org.apache.beam.sdk.state.TimerSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+import org.apache.beam.sdk.transforms.Values;
+import org.apache.beam.sdk.transforms.WithKeys;
+import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollection.IsBounded;
+import org.apache.beam.sdk.values.TypeDescriptor;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+
+/**
+ * Takes FeatureRow, and merges them if they have the same FeatureRowKey, so that the latest values
+ * will be emitted. It emits only once for batch.
+ *
+ * For streaming we emits after a delay of 10 seconds (event time) by default we keep the
+ * previous state around for merging with future events. These timeout after 30 minutes by default.
+ */
+public class CoalesceFeatureRows
+ extends PTransform, PCollection> {
+
+ private static final SerializableFunction KEY_FUNCTION =
+ (row) ->
+ CoalesceKey.newBuilder()
+ .setEntityName(row.getEntityName())
+ .setEntityKey(row.getEntityKey())
+ .build();
+
+ private static final Duration DEFAULT_DELAY = Duration.standardSeconds(10);
+ private static final Duration DEFAULT_TIMEOUT = Duration.ZERO;
+
+ private Duration delay;
+ private Duration timeout;
+
+ CoalesceFeatureRows() {
+ this(0, 0);
+ }
+
+ public CoalesceFeatureRows(long delaySeconds, long timeoutSeconds) {
+ this(Duration.standardSeconds(delaySeconds), Duration.standardSeconds(timeoutSeconds));
+ }
+
+ public CoalesceFeatureRows(Duration delay, Duration timeout) {
+ this.delay = (delay.isEqual(Duration.ZERO)) ? DEFAULT_DELAY : delay;
+ this.timeout = (timeout.isEqual(Duration.ZERO)) ? DEFAULT_TIMEOUT : timeout;
+ }
+
+ /** Return a FeatureRow of the new features accumulated since the given timestamp */
+ public static FeatureRow toFeatureRow(CoalesceAccum accum, long counter) {
+ Preconditions.checkArgument(
+ counter <= accum.getCounter(),
+ "Accumulator has no features at or newer than the provided counter");
+ FeatureRow.Builder builder =
+ FeatureRow.newBuilder()
+ .setEntityName(accum.getEntityName())
+ .setEntityKey(accum.getEntityKey());
+ // This will be the latest timestamp
+ if (accum.hasEventTimestamp()) {
+ builder.setEventTimestamp(accum.getEventTimestamp());
+ }
+
+ Map features = accum.getFeaturesMap();
+ if (counter <= 0) {
+ builder.addAllFeatures(features.values());
+ } else {
+ List featureList =
+ accum
+ .getFeatureMarksMap()
+ .entrySet()
+ .stream()
+ .filter((e) -> e.getValue() > counter)
+ .map((e) -> features.get(e.getKey()))
+ .collect(Collectors.toList());
+ builder.addAllFeatures(featureList);
+ }
+ return builder.build();
+ }
+
+ public static FeatureRow combineFeatureRows(Iterable rows) {
+ return toFeatureRow(combineFeatureRows(CoalesceAccum.getDefaultInstance(), rows), 0);
+ }
+
+ public static CoalesceAccum combineFeatureRows(CoalesceAccum seed, Iterable rows) {
+ CoalesceAccum.Builder accum = seed.toBuilder();
+ Map features = new HashMap<>();
+ Map featureMarks = new HashMap<>();
+ long rowCount = seed.getCounter();
+ for (FeatureRow row : rows) {
+ rowCount += 1;
+ if (Timestamps.compare(accum.getEventTimestamp(), row.getEventTimestamp()) <= 0) {
+ // row has later timestamp than accum.
+ for (Feature feature : row.getFeaturesList()) {
+ features.put(feature.getId(), feature);
+ // These marks are used to determine which features are new when we convert an accum
+ // back into a FeatureRow.
+ featureMarks.put(feature.getId(), rowCount);
+ }
+ accum.setEntityName(row.getEntityName());
+ accum.setEntityKey(row.getEntityKey());
+ if (row.hasEventTimestamp()) {
+ accum.setEventTimestamp(row.getEventTimestamp());
+ }
+ } else {
+ for (Feature feature : row.getFeaturesList()) {
+ String featureId = feature.getId();
+ // only insert an older feature if there was no newer one.
+ if (!features.containsKey(featureId)) {
+ features.put(featureId, feature);
+ featureMarks.put(feature.getId(), rowCount);
+ }
+ }
+ }
+ }
+ if (rowCount == seed.getCounter()) {
+ return seed;
+ } else {
+ return accum
+ .setCounter(rowCount)
+ .putAllFeatures(features)
+ .putAllFeatureMarks(featureMarks)
+ .build();
+ }
+ }
+
+ @Override
+ public PCollection expand(PCollection input) {
+ PCollection> kvs =
+ input
+ .apply(WithKeys.of(KEY_FUNCTION).withKeyType(TypeDescriptor.of(CoalesceKey.class)))
+ .setCoder(
+ KvCoder.of(ProtoCoder.of(CoalesceKey.class), ProtoCoder.of(FeatureRow.class)));
+
+ if (kvs.isBounded().equals(IsBounded.UNBOUNDED)) {
+ return kvs.apply(
+ "Configure window",
+ Window.>configure()
+ .withAllowedLateness(Duration.ZERO)
+ .discardingFiredPanes()
+ .triggering(AfterProcessingTime.pastFirstElementInPane()))
+ .apply(ParDo.of(new CombineStateDoFn(delay, timeout)))
+ .apply(Values.create());
+ } else {
+ return kvs.apply(Combine.perKey(CoalesceFeatureRows::combineFeatureRows))
+ .apply(Values.create());
+ }
+ }
+
+ @Slf4j
+ @AllArgsConstructor
+ public static class CombineStateDoFn
+ extends DoFn, KV> {
+
+ @StateId("lastKnownAccumValue")
+ private final StateSpec> lastKnownAccumValueSpecs =
+ StateSpecs.value(ProtoCoder.of(CoalesceAccum.class));
+
+ @StateId("newElementsBag")
+ private final StateSpec> newElementsBag =
+ StateSpecs.bag(ProtoCoder.of(FeatureRow.class));
+
+ @StateId("lastTimerTimestamp")
+ private final StateSpec> lastTimerTimestamp = StateSpecs.value();
+
+ @TimerId("bufferTimer")
+ private final TimerSpec bufferTimer = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ @TimerId("timeoutTimer")
+ private final TimerSpec timeoutTimer = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private Duration delay;
+ private Duration timeout;
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext context,
+ @StateId("newElementsBag") BagState newElementsBag,
+ @TimerId("bufferTimer") Timer bufferTimer,
+ @TimerId("timeoutTimer") Timer timeoutTimer,
+ @StateId("lastTimerTimestamp") ValueState lastTimerTimestampValue) {
+ newElementsBag.add(context.element().getValue());
+ log.debug("Adding FeatureRow to state bag {}", context.element());
+
+ Instant lastTimerTimestamp = lastTimerTimestampValue.read();
+ Instant contextTimestamp = context.timestamp();
+ if (lastTimerTimestamp == null && timeout.isLongerThan(Duration.ZERO)) {
+ // We never timeout the state if a timeout is not set.
+ timeoutTimer.offset(timeout).setRelative();
+ }
+ if (lastTimerTimestamp == null
+ || lastTimerTimestamp.isBefore(contextTimestamp)
+ || lastTimerTimestamp.equals(contextTimestamp)) {
+ lastTimerTimestamp = context.timestamp().plus(delay);
+ log.debug("Setting timer for key {} to {}", context.element().getKey(), lastTimerTimestamp);
+ lastTimerTimestampValue.write(lastTimerTimestamp);
+ bufferTimer.offset(delay).setRelative();
+ }
+ }
+
+ @OnTimer("bufferTimer")
+ public void bufferOnTimer(
+ OnTimerContext context,
+ OutputReceiver> out,
+ @StateId("newElementsBag") BagState newElementsBag,
+ @StateId("lastKnownAccumValue") ValueState lastKnownAccumValue) {
+ log.debug("bufferOnTimer triggered {}", context.timestamp());
+ flush(out, newElementsBag, lastKnownAccumValue);
+ }
+
+ @OnTimer("timeoutTimer")
+ public void timeoutOnTimer(
+ OnTimerContext context,
+ OutputReceiver> out,
+ @StateId("newElementsBag") BagState newElementsBag,
+ @StateId("lastKnownAccumValue") ValueState lastKnownAccumValue) {
+ log.debug("timeoutOnTimer triggered {}", context.timestamp());
+ flush(out, newElementsBag, lastKnownAccumValue);
+ newElementsBag.clear();
+ lastKnownAccumValue.clear();
+ }
+
+ public void flush(
+ OutputReceiver> out,
+ @StateId("newElementsBag") BagState newElementsBag,
+ @StateId("lastKnownAccumValue") ValueState lastKnownAccumValue) {
+ log.debug("Flush triggered");
+ Iterable rows = newElementsBag.read();
+ if (!rows.iterator().hasNext()) {
+ log.debug("Flush with no new elements");
+ return;
+ }
+ CoalesceAccum lastKnownAccum = lastKnownAccumValue.read();
+ if (lastKnownAccum == null) {
+ lastKnownAccum = CoalesceAccum.getDefaultInstance();
+ }
+ // Check if we have more than one value in our list.
+ CoalesceAccum accum = combineFeatureRows(lastKnownAccum, rows);
+ FeatureRow row = toFeatureRow(accum, lastKnownAccum.getCounter());
+ log.debug("Timer fired and added FeatureRow to output {}", row);
+ // Clear the elements now that they have been processed
+ newElementsBag.clear();
+ lastKnownAccumValue.write(accum);
+
+ // Output the value stored in the the processed que which matches this timers time
+ out.output(KV.of(KEY_FUNCTION.apply(row), row));
+ }
+ }
+}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/CoalescePFeatureRows.java b/ingestion/src/main/java/feast/ingestion/transform/CoalescePFeatureRows.java
new file mode 100644
index 00000000000..cf7932f0155
--- /dev/null
+++ b/ingestion/src/main/java/feast/ingestion/transform/CoalescePFeatureRows.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2019 The Feast Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package feast.ingestion.transform;
+
+import feast.ingestion.values.PFeatureRows;
+import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
+import feast.types.FeatureRowProto.FeatureRow;
+import lombok.AllArgsConstructor;
+import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptor;
+
+/**
+ * This class is a work around to make some refactoring easier. PFeatureRows should be deprecated.
+ */
+@AllArgsConstructor
+public class CoalescePFeatureRows extends
+ PTransform {
+
+ private long delaySeconds;
+ private long timeoutSeconds;
+
+ @Override
+ public PFeatureRows expand(PFeatureRows input) {
+ PCollection output = input.getMain()
+ .apply(MapElements.into(TypeDescriptor.of(FeatureRow.class))
+ .via(FeatureRowExtended::getRow))
+ .apply(new CoalesceFeatureRows(delaySeconds, timeoutSeconds))
+ .apply(new ToFeatureRowExtended());
+ return PFeatureRows.of(output, input.getErrors());
+ }
+}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/ErrorsStoreTransform.java b/ingestion/src/main/java/feast/ingestion/transform/ErrorsStoreTransform.java
index 97799a0ae86..70f7f2d3c75 100644
--- a/ingestion/src/main/java/feast/ingestion/transform/ErrorsStoreTransform.java
+++ b/ingestion/src/main/java/feast/ingestion/transform/ErrorsStoreTransform.java
@@ -18,63 +18,61 @@
package feast.ingestion.transform;
import static com.google.common.base.Preconditions.checkArgument;
-import static feast.ingestion.util.JsonUtil.convertJsonStringToMap;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
import com.google.inject.Inject;
import feast.ingestion.model.Specs;
-import feast.ingestion.options.ImportJobOptions;
-import feast.ingestion.transform.FeatureIO.Write;
+import feast.ingestion.options.ImportJobPipelineOptions;
+import feast.ingestion.util.PathUtil;
import feast.specs.StorageSpecProto.StorageSpec;
-import feast.storage.ErrorsStore;
-import feast.storage.noop.NoOpIO;
+import feast.store.FeatureStoreWrite;
+import feast.store.errors.FeatureErrorsFactory;
+import feast.store.errors.json.JsonFileErrorsFactory;
import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
+import java.nio.file.Path;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PDone;
-import org.apache.hadoop.hbase.util.Strings;
@Slf4j
-public class ErrorsStoreTransform extends FeatureIO.Write {
+public class ErrorsStoreTransform extends FeatureStoreWrite {
- private String errorsStoreType;
- private StorageSpec errorsStoreSpec;
+ private String workspace;
private Specs specs;
- private List errorsStores;
+ private List errorsStoreFactories;
@Inject
public ErrorsStoreTransform(
- ImportJobOptions options, Specs specs, List errorsStores) {
+ ImportJobPipelineOptions options, Specs specs,
+ List errorsStoreFactories) {
+ this.workspace = options.getWorkspace();
this.specs = specs;
- this.errorsStores = errorsStores;
- this.errorsStoreType = options.getErrorsStoreType();
-
- if (!Strings.isEmpty(errorsStoreType)) {
- this.errorsStoreSpec =
- StorageSpec.newBuilder()
- .setType(errorsStoreType)
- .putAllOptions(convertJsonStringToMap(options.getErrorsStoreOptions()))
- .build();
- }
+ this.errorsStoreFactories = errorsStoreFactories;
}
@Override
public PDone expand(PCollection input) {
- Write write;
- if (Strings.isEmpty(errorsStoreType)) {
- write = new NoOpIO.Write();
- } else {
- write = getErrorStore().create(this.errorsStoreSpec, specs);
+ StorageSpec errorsStoreSpec = specs.getErrorsStoreSpec();
+ if (Strings.isNullOrEmpty(errorsStoreSpec.getType())) {
+ Preconditions.checkArgument(!Strings.isNullOrEmpty(workspace), "workspace must be provided");
+ Path workspaceErrorsPath = PathUtil.getPath(workspace).resolve("errors");
+ errorsStoreSpec = StorageSpec.newBuilder()
+ .setId("workspace/errors")
+ .setType(JsonFileErrorsFactory.JSON_FILES_TYPE)
+ .putOptions("path", workspaceErrorsPath.toString()).build();
}
- input.apply("errors to " + String.valueOf(errorsStoreType), write);
+ input.apply("Write errors" + errorsStoreSpec.getType(),
+ getErrorStore(errorsStoreSpec.getType()).create(errorsStoreSpec, specs));
return PDone.in(input.getPipeline());
}
- ErrorsStore getErrorStore() {
- checkArgument(!errorsStoreType.isEmpty(), "Errors store type not provided");
- for (ErrorsStore errorsStore : errorsStores) {
- if (errorsStore.getType().equals(errorsStoreType)) {
- return errorsStore;
+ FeatureErrorsFactory getErrorStore(String type) {
+ checkArgument(!type.isEmpty(), "Errors store type not provided");
+ for (FeatureErrorsFactory errorsStoreFactory : errorsStoreFactories) {
+ if (errorsStoreFactory.getType().equals(type)) {
+ return errorsStoreFactory;
}
}
throw new IllegalArgumentException("Errors store type not found");
diff --git a/ingestion/src/main/java/feast/ingestion/transform/FeatureIO.java b/ingestion/src/main/java/feast/ingestion/transform/FeatureIO.java
deleted file mode 100644
index c6d9a3b1c09..00000000000
--- a/ingestion/src/main/java/feast/ingestion/transform/FeatureIO.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Copyright 2018 The Feast Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package feast.ingestion.transform;
-
-import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
-import feast.types.FeatureRowProto.FeatureRow;
-import lombok.AllArgsConstructor;
-import org.apache.beam.sdk.transforms.PTransform;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PDone;
-import org.apache.beam.sdk.values.PInput;
-
-public class FeatureIO {
-
- public abstract static class Read extends PTransform> {}
-
- public abstract static class Write extends PTransform, PDone> {}
-
- /** Used during setup if read transform can not be determined. */
- @AllArgsConstructor
- public static class UnknownRead extends Read {
- private String message;
-
- @Override
- public PCollection expand(PInput input) {
- throw new IllegalArgumentException(message);
- }
- }
-}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/ReadFeaturesTransform.java b/ingestion/src/main/java/feast/ingestion/transform/ReadFeaturesTransform.java
index 5da8f6e7532..15dc223cc2f 100644
--- a/ingestion/src/main/java/feast/ingestion/transform/ReadFeaturesTransform.java
+++ b/ingestion/src/main/java/feast/ingestion/transform/ReadFeaturesTransform.java
@@ -19,6 +19,7 @@
import com.google.common.base.Preconditions;
import com.google.inject.Inject;
+import feast.ingestion.model.Specs;
import feast.source.FeatureSourceFactory;
import feast.source.FeatureSourceFactoryService;
import feast.specs.ImportSpecProto.ImportSpec;
@@ -32,8 +33,8 @@ public class ReadFeaturesTransform extends PTransform {
- private List stores;
+ private List stores;
private Specs specs;
@Inject
- public ServingStoreTransform(List stores, Specs specs) {
+ public ServingStoreTransform(List stores, Specs specs) {
this.stores = stores;
this.specs = specs;
}
@@ -46,11 +45,13 @@ public PFeatureRows expand(PFeatureRows input) {
input.apply(
"Split to serving stores",
new SplitOutputByStore(
- stores, (featureSpec) -> featureSpec.getDataStores().getServing().getId(), specs));
+ stores, (featureSpec) -> featureSpec.getDataStores().getServing().getId(), specs,
+ specs.getServingStorageSpecs()));
output.getMain().apply("metrics.store.lag", ParDo.of(FeastMetrics.lagUpdateDoFn()));
output.getMain().apply("metrics.store.main", ParDo.of(FeastMetrics.incrDoFn("serving_stored")));
- output.getErrors().apply("metrics.store.errors", ParDo.of(FeastMetrics.incrDoFn("serving_errors")));
+ output.getErrors()
+ .apply("metrics.store.errors", ParDo.of(FeastMetrics.incrDoFn("serving_errors")));
return output;
}
}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/SplitOutputByStore.java b/ingestion/src/main/java/feast/ingestion/transform/SplitOutputByStore.java
index cb11490dd35..ffb1393698e 100644
--- a/ingestion/src/main/java/feast/ingestion/transform/SplitOutputByStore.java
+++ b/ingestion/src/main/java/feast/ingestion/transform/SplitOutputByStore.java
@@ -20,12 +20,12 @@
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import feast.ingestion.model.Specs;
-import feast.ingestion.transform.FeatureIO.Write;
import feast.ingestion.transform.SplitFeatures.MultiOutputSplit;
import feast.ingestion.values.PFeatureRows;
import feast.specs.FeatureSpecProto.FeatureSpec;
import feast.specs.StorageSpecProto.StorageSpec;
-import feast.storage.FeatureStore;
+import feast.store.FeatureStoreFactory;
+import feast.store.FeatureStoreWrite;
import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
import java.util.Collection;
import java.util.HashMap;
@@ -46,20 +46,21 @@
@Slf4j
public class SplitOutputByStore extends PTransform {
- private Collection stores;
+ private Collection stores;
private SerializableFunction selector;
private Specs specs;
+ private Map storageSpecs;
@Override
public PFeatureRows expand(PFeatureRows input) {
- Map transforms = getFeatureStoreTransforms();
+ Map transforms = getFeatureStoreTransforms();
Set keys = transforms.keySet();
log.info(String.format("Splitting on keys = [%s]", String.join(",", keys)));
MultiOutputSplit splitter = new MultiOutputSplit<>(selector, keys, specs);
PCollectionTuple splits = input.getMain().apply(splitter);
- Map, Write> taggedTransforms = new HashMap<>();
+ Map, FeatureStoreWrite> taggedTransforms = new HashMap<>();
for (String key : transforms.keySet()) {
TupleTag tag = splitter.getStrategy().getTag(key);
taggedTransforms.put(tag, transforms.get(key));
@@ -72,18 +73,17 @@ public PFeatureRows expand(PFeatureRows input) {
input.getErrors());
}
- private Map getStoresMap() {
- Map storesMap = new HashMap<>();
- for (FeatureStore servingStore : stores) {
+ private Map getStoresMap() {
+ Map storesMap = new HashMap<>();
+ for (FeatureStoreFactory servingStore : stores) {
storesMap.put(servingStore.getType(), servingStore);
}
return storesMap;
}
- private Map getFeatureStoreTransforms() {
- Map storesMap = getStoresMap();
- Map transforms = new HashMap<>();
- Map storageSpecs = specs.getStorageSpecs();
+ private Map getFeatureStoreTransforms() {
+ Map storesMap = getStoresMap();
+ Map transforms = new HashMap<>();
for (String storeId : storageSpecs.keySet()) {
StorageSpec storageSpec = storageSpecs.get(storeId);
String type = storageSpec.getType();
@@ -109,14 +109,14 @@ private Map getFeatureStoreTransforms() {
public static class WriteTags extends
PTransform> {
- private Map, Write> transforms;
+ private Map, FeatureStoreWrite> transforms;
private TupleTag mainTag;
@Override
public PCollection expand(PCollectionTuple tuple) {
List> outputList = Lists.newArrayList();
for (TupleTag tag : transforms.keySet()) {
- Write write = transforms.get(tag);
+ FeatureStoreWrite write = transforms.get(tag);
Preconditions.checkNotNull(write, String.format("Null transform for tag=%s", tag.getId()));
PCollection input = tuple.get(tag);
input.apply(String.format("Write to %s", tag.getId()), write);
diff --git a/ingestion/src/main/java/feast/ingestion/transform/WarehouseStoreTransform.java b/ingestion/src/main/java/feast/ingestion/transform/WarehouseStoreTransform.java
index dcfb9c44334..90e8211f2a2 100644
--- a/ingestion/src/main/java/feast/ingestion/transform/WarehouseStoreTransform.java
+++ b/ingestion/src/main/java/feast/ingestion/transform/WarehouseStoreTransform.java
@@ -15,26 +15,27 @@
*
*/
+
package feast.ingestion.transform;
import com.google.inject.Inject;
+import feast.ingestion.metrics.FeastMetrics;
+import feast.ingestion.model.Specs;
+import feast.ingestion.values.PFeatureRows;
+import feast.store.warehouse.FeatureWarehouseFactory;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
-import feast.ingestion.metrics.FeastMetrics;
-import feast.ingestion.model.Specs;
-import feast.ingestion.values.PFeatureRows;
-import feast.storage.WarehouseStore;
@Slf4j
public class WarehouseStoreTransform extends PTransform {
- private List stores;
+ private List stores;
private Specs specs;
@Inject
- public WarehouseStoreTransform(List stores, Specs specs) {
+ public WarehouseStoreTransform(List stores, Specs specs) {
this.stores = stores;
this.specs = specs;
}
@@ -47,9 +48,12 @@ public PFeatureRows expand(PFeatureRows input) {
new SplitOutputByStore(
stores,
(featureSpec) -> featureSpec.getDataStores().getWarehouse().getId(),
- specs));
- output.getMain().apply("metrics.store.main", ParDo.of(FeastMetrics.incrDoFn("warehouse_stored")));
- output.getErrors().apply("metrics.store.errors", ParDo.of(FeastMetrics.incrDoFn("warehouse_errors")));
+ specs,
+ specs.getWarehouseStorageSpecs()));
+ output.getMain()
+ .apply("metrics.store.main", ParDo.of(FeastMetrics.incrDoFn("warehouse_stored")));
+ output.getErrors()
+ .apply("metrics.store.errors", ParDo.of(FeastMetrics.incrDoFn("warehouse_errors")));
return output;
}
}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/fn/ConvertTypesDoFn.java b/ingestion/src/main/java/feast/ingestion/transform/fn/ConvertTypesDoFn.java
index c26b25a81fb..c5ed560eeaf 100644
--- a/ingestion/src/main/java/feast/ingestion/transform/fn/ConvertTypesDoFn.java
+++ b/ingestion/src/main/java/feast/ingestion/transform/fn/ConvertTypesDoFn.java
@@ -36,14 +36,15 @@ public void processElementImpl(ProcessContext context) {
FeatureRow.Builder rowBuilder = FeatureRow.newBuilder();
rowBuilder
.setEntityName(row.getEntityName())
- .setEventTimestamp(row.getEventTimestamp())
.setEntityKey(row.getEntityKey());
+ if (row.hasEventTimestamp()) {
+ rowBuilder.setEventTimestamp(row.getEventTimestamp());
+ }
for (Feature feature : row.getFeaturesList()) {
String featureId = feature.getId();
FeatureSpec featureSpec = specs.getFeatureSpec(featureId);
- rowBuilder.setGranularity(featureSpec.getGranularity());
rowBuilder.addFeatures(
Feature.newBuilder()
.setId(featureId)
diff --git a/ingestion/src/main/java/feast/ingestion/transform/fn/LoggerDoFn.java b/ingestion/src/main/java/feast/ingestion/transform/fn/LoggerDoFn.java
index 8811faa173c..d641c690a29 100644
--- a/ingestion/src/main/java/feast/ingestion/transform/fn/LoggerDoFn.java
+++ b/ingestion/src/main/java/feast/ingestion/transform/fn/LoggerDoFn.java
@@ -17,7 +17,6 @@
package feast.ingestion.transform.fn;
-import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.Message;
import com.google.protobuf.util.JsonFormat;
import com.google.protobuf.util.JsonFormat.Printer;
@@ -46,7 +45,7 @@ public void processElement(ProcessContext context) {
String message;
try {
message = prefix + printer.print(context.element());
- } catch (InvalidProtocolBufferException e) {
+ } catch (Exception e) {
log.error(e.getMessage(), e);
message = prefix + context.element().toString();
}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/fn/RoundEventTimestampsDoFn.java b/ingestion/src/main/java/feast/ingestion/transform/fn/RoundEventTimestampsDoFn.java
deleted file mode 100644
index 7170e65915b..00000000000
--- a/ingestion/src/main/java/feast/ingestion/transform/fn/RoundEventTimestampsDoFn.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * Copyright 2018 The Feast Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package feast.ingestion.transform.fn;
-
-import org.apache.beam.sdk.transforms.DoFn;
-import feast.ingestion.util.DateUtil;
-import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
-import feast.types.FeatureRowProto.FeatureRow;
-
-public class RoundEventTimestampsDoFn extends DoFn {
- @ProcessElement
- public void processElement(ProcessContext context) {
- FeatureRowExtended rowExtended = context.element();
- FeatureRow row = rowExtended.getRow();
- com.google.protobuf.Timestamp timestamp = rowExtended.getRow().getEventTimestamp();
- row =
- row.toBuilder()
- .setEventTimestamp(DateUtil.roundToGranularity(timestamp, row.getGranularity()))
- .build();
- context.output(rowExtended.toBuilder().setRow(row).build());
- }
-}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/fn/SplitFeaturesDoFn.java b/ingestion/src/main/java/feast/ingestion/transform/fn/SplitFeaturesDoFn.java
index 8fdb226029e..5c46a5dd5df 100644
--- a/ingestion/src/main/java/feast/ingestion/transform/fn/SplitFeaturesDoFn.java
+++ b/ingestion/src/main/java/feast/ingestion/transform/fn/SplitFeaturesDoFn.java
@@ -50,7 +50,6 @@ public void processElement(ProcessContext context) {
if (builder == null) {
builder =
FeatureRow.newBuilder()
- .setGranularity(row.getGranularity())
.setEventTimestamp(row.getEventTimestamp())
.setEntityName(row.getEntityName())
.setEntityKey(row.getEntityKey());
diff --git a/ingestion/src/main/java/feast/ingestion/transform/fn/ValidateFeatureRowsDoFn.java b/ingestion/src/main/java/feast/ingestion/transform/fn/ValidateFeatureRowsDoFn.java
index 40fc6054fd8..67aec7d255a 100644
--- a/ingestion/src/main/java/feast/ingestion/transform/fn/ValidateFeatureRowsDoFn.java
+++ b/ingestion/src/main/java/feast/ingestion/transform/fn/ValidateFeatureRowsDoFn.java
@@ -22,6 +22,7 @@
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
+import com.google.protobuf.util.Timestamps;
import feast.ingestion.exceptions.ValidationException;
import feast.ingestion.metrics.FeastMetrics;
import feast.ingestion.model.Specs;
@@ -31,13 +32,12 @@
import feast.specs.ImportSpecProto.Field;
import feast.specs.ImportSpecProto.ImportSpec;
import feast.specs.StorageSpecProto.StorageSpec;
-import feast.storage.ServingStore;
-import feast.storage.WarehouseStore;
-import feast.storage.service.ServingStoreService;
-import feast.storage.service.WarehouseStoreService;
+import feast.store.serving.FeatureServingFactory;
+import feast.store.serving.FeatureServingFactoryService;
+import feast.store.warehouse.FeatureWarehouseFactory;
+import feast.store.warehouse.FeatureWarehouseFactoryService;
import feast.types.FeatureProto.Feature;
import feast.types.FeatureRowProto.FeatureRow;
-import feast.types.GranularityProto.Granularity.Enum;
import feast.types.ValueProto.ValueType;
import java.util.ArrayList;
import java.util.HashSet;
@@ -66,10 +66,10 @@ public void setup() {
featureIds.add(field.getFeatureId());
}
}
- for (ServingStore store : ServingStoreService.getAll()) {
+ for (FeatureServingFactory store : FeatureServingFactoryService.getAll()) {
supportedServingTypes.add(store.getType());
}
- for (WarehouseStore store : WarehouseStoreService.getAll()) {
+ for (FeatureWarehouseFactory store : FeatureWarehouseFactoryService.getAll()) {
supportedWarehouseTypes.add(store.getType());
}
}
@@ -79,7 +79,6 @@ public void processElementImpl(ProcessContext context) {
FeatureRow row = context.element().getRow();
EntitySpec entitySpec = specs.getEntitySpec(row.getEntityName());
Preconditions.checkNotNull(entitySpec, "Entity spec not found for " + row.getEntityName());
- ImportSpec importSpec = specs.getImportSpec();
try {
checkArgument(!row.getEntityKey().isEmpty(), "Entity key must not be empty");
@@ -90,11 +89,9 @@ public void processElementImpl(ProcessContext context) {
String.format(
"Row entity not found in import spec entities. entity=%s", row.getEntityName()));
- checkArgument(
- !row.getGranularity().equals(Enum.UNRECOGNIZED),
- String.format("Unrecognised granularity %s", row.getGranularity()));
-
checkArgument(row.hasEventTimestamp(), "Must have eventTimestamp set");
+ Timestamps.checkValid(row.getEventTimestamp());
+
checkArgument(row.getFeaturesCount() > 0, "Must have at least one feature set");
for (Feature feature : row.getFeaturesList()) {
@@ -102,19 +99,20 @@ public void processElementImpl(ProcessContext context) {
checkNotNull(
featureSpec, String.format("Feature spec not found featureId=%s", feature.getId()));
- String storageStoreId = featureSpec.getDataStores().getServing().getId();
- StorageSpec servingStorageSpec = specs.getStorageSpec(storageStoreId);
- checkArgument(
- supportedServingTypes.contains(servingStorageSpec.getType()),
- String.format("Serving storage type=%s not supported", servingStorageSpec.getType()));
+ String servingStoreId = featureSpec.getDataStores().getServing().getId();
+ if (!servingStoreId.equals(EMPTY_STORE)) {
+ StorageSpec servingStorageSpec = specs.getServingStorageSpecs()
+ .getOrDefault(servingStoreId, null);
+ checkNotNull(servingStorageSpec,
+ "Serving storage specs not found for store id " + servingStoreId);
+ }
String warehouseStoreId = featureSpec.getDataStores().getWarehouse().getId();
if (!warehouseStoreId.equals(EMPTY_STORE)) {
- StorageSpec warehouseStorageSpec = specs.getStorageSpec(warehouseStoreId);
- checkArgument(
- supportedWarehouseTypes.contains(warehouseStorageSpec.getType()),
- String.format(
- "Warehouse storage type=%s not supported", servingStorageSpec.getType()));
+ StorageSpec warehouseStorageSpec = specs.getWarehouseStorageSpecs()
+ .getOrDefault(warehouseStoreId, null);
+ checkNotNull(warehouseStorageSpec,
+ "Warehouse storage specs not found for store id " + servingStoreId);
}
checkArgument(
@@ -123,11 +121,6 @@ public void processElementImpl(ProcessContext context) {
"Feature must have same entity as row. featureId=%s FeatureRow.entityName=%s FeatureSpec.entity=%s",
feature.getId(), row.getEntityName(), featureSpec.getEntity()));
- checkArgument(
- featureSpec.getGranularity().equals(row.getGranularity()),
- String.format(
- "Feature must have same granularity as entity, featureId=%s", feature.getId()));
-
ValueType.Enum expectedType = featureSpec.getValueType();
ValueType.Enum actualType = Values.toValueType(feature.getValue());
checkArgument(
diff --git a/ingestion/src/main/java/feast/ingestion/util/DateUtil.java b/ingestion/src/main/java/feast/ingestion/util/DateUtil.java
index 15c4cc4d465..ad779dbea33 100644
--- a/ingestion/src/main/java/feast/ingestion/util/DateUtil.java
+++ b/ingestion/src/main/java/feast/ingestion/util/DateUtil.java
@@ -20,11 +20,12 @@
import com.google.protobuf.Timestamp;
import java.time.Instant;
import org.joda.time.DateTime;
-import org.joda.time.DateTimeField;
import org.joda.time.DateTimeZone;
-import org.joda.time.MutableDateTime;
-import org.joda.time.format.*;
-import feast.types.GranularityProto.Granularity;
+import org.joda.time.format.DateTimeFormat;
+import org.joda.time.format.DateTimeFormatter;
+import org.joda.time.format.DateTimeFormatterBuilder;
+import org.joda.time.format.DateTimeParser;
+import org.joda.time.format.ISODateTimeFormat;
public class DateUtil {
@@ -34,12 +35,12 @@ public class DateUtil {
DateTimeFormatterBuilder formatterBuilder = new DateTimeFormatterBuilder();
DateTimeFormatter base = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss");
DateTimeFormatter zone = DateTimeFormat.forPattern(" ZZZ");
- DateTimeParser fractionSecondParser = new DateTimeFormatterBuilder()
- .appendLiteral(".")
- .appendFractionOfSecond(1,6)
- .toParser();
+ DateTimeParser fractionSecondParser =
+ new DateTimeFormatterBuilder().appendLiteral(".").appendFractionOfSecond(1, 6).toParser();
- FALLBACK_TIMESTAMP_FORMAT = formatterBuilder.append(base)
+ FALLBACK_TIMESTAMP_FORMAT =
+ formatterBuilder
+ .append(base)
.appendOptional(fractionSecondParser)
.append(zone)
.toFormatter();
@@ -93,30 +94,4 @@ public static Timestamp maxTimestamp(Timestamp a, Timestamp b) {
public static long toMillis(Timestamp timestamp) {
return toDateTime(timestamp).getMillis();
}
-
- public static Timestamp roundToGranularity(Timestamp timestamp, Granularity.Enum granularity) {
- MutableDateTime dt = new MutableDateTime(DateTimeZone.UTC);
- DateTimeField roundingField;
- switch (granularity) {
- case DAY:
- roundingField = dt.getChronology().dayOfMonth();
- break;
- case HOUR:
- roundingField = dt.getChronology().hourOfDay();
- break;
- case MINUTE:
- roundingField = dt.getChronology().minuteOfHour();
- break;
- case SECOND:
- roundingField = dt.getChronology().secondOfMinute();
- break;
- case NONE:
- return Timestamp.newBuilder().setSeconds(0).setNanos(0).build();
- default:
- throw new RuntimeException("Unrecognised time series granularity");
- }
- dt.setRounding(roundingField, MutableDateTime.ROUND_FLOOR);
- dt.setMillis(toDateTime(timestamp).getMillis());
- return toTimestamp(dt.toDateTime());
- }
}
diff --git a/ingestion/src/main/java/feast/ingestion/util/ProtoUtil.java b/ingestion/src/main/java/feast/ingestion/util/ProtoUtil.java
index de59dac3818..9b71faff276 100644
--- a/ingestion/src/main/java/feast/ingestion/util/ProtoUtil.java
+++ b/ingestion/src/main/java/feast/ingestion/util/ProtoUtil.java
@@ -26,18 +26,18 @@
import com.google.protobuf.Message;
import com.google.protobuf.util.JsonFormat;
import java.io.IOException;
+import java.nio.file.Files;
import java.nio.file.Path;
public class ProtoUtil {
- private ProtoUtil() {}
+
+ private ProtoUtil() {
+ }
public static T decodeProtoYamlFile(Path path, T prototype)
throws IOException {
- ObjectMapper yamlMapper = new ObjectMapper(new YAMLFactory());
- ObjectMap map = yamlMapper.readerFor(ObjectMap.class).readValue(path.toFile());
- ObjectMapper jsonMapper = new ObjectMapper(new JsonFactory());
- String json = jsonMapper.writerFor(ObjectMap.class).writeValueAsString(map);
- return decodeProtoJson(json, prototype);
+ String yaml = String.join("\n", Files.readAllLines(path));
+ return decodeProtoYaml(yaml, prototype);
}
public static T decodeProtoYaml(String yamlString, T prototype)
diff --git a/ingestion/src/main/java/feast/ingestion/values/PFeatureRows.java b/ingestion/src/main/java/feast/ingestion/values/PFeatureRows.java
index 3c46e92f158..86a710ddae6 100644
--- a/ingestion/src/main/java/feast/ingestion/values/PFeatureRows.java
+++ b/ingestion/src/main/java/feast/ingestion/values/PFeatureRows.java
@@ -14,7 +14,6 @@
* limitations under the License.
*
*/
-
package feast.ingestion.values;
import feast.ingestion.transform.fn.BaseFeatureDoFn;
diff --git a/ingestion/src/main/java/feast/options/OptionsParser.java b/ingestion/src/main/java/feast/options/OptionsParser.java
index 7eab269637d..aca61d7f144 100644
--- a/ingestion/src/main/java/feast/options/OptionsParser.java
+++ b/ingestion/src/main/java/feast/options/OptionsParser.java
@@ -22,7 +22,6 @@
import com.fasterxml.jackson.module.jsonSchema.JsonSchema;
import com.fasterxml.jackson.module.jsonSchema.JsonSchemaGenerator;
import com.google.common.collect.Lists;
-import feast.source.csv.CsvFileFeatureSource.CsvFileFeatureSourceOptions;
import java.io.IOException;
import java.util.List;
import java.util.Map;
diff --git a/ingestion/src/main/java/feast/source/bigquery/BigQueryFeatureSource.java b/ingestion/src/main/java/feast/source/bigquery/BigQueryFeatureSource.java
index 9ce7f965f2c..b183ffbddc0 100644
--- a/ingestion/src/main/java/feast/source/bigquery/BigQueryFeatureSource.java
+++ b/ingestion/src/main/java/feast/source/bigquery/BigQueryFeatureSource.java
@@ -65,7 +65,7 @@ public class BigQueryFeatureSource extends FeatureSource {
@Override
public PCollection expand(PInput input) {
BigQuerySourceOptions options = OptionsParser
- .parse(importSpec.getOptionsMap(), BigQuerySourceOptions.class);
+ .parse(importSpec.getSourceOptionsMap(), BigQuerySourceOptions.class);
List entities = importSpec.getEntitiesList();
Preconditions.checkArgument(
diff --git a/ingestion/src/main/java/feast/source/bigquery/BigQueryToFeatureRowFn.java b/ingestion/src/main/java/feast/source/bigquery/BigQueryToFeatureRowFn.java
index 795735410d9..dfe26c45b77 100644
--- a/ingestion/src/main/java/feast/source/bigquery/BigQueryToFeatureRowFn.java
+++ b/ingestion/src/main/java/feast/source/bigquery/BigQueryToFeatureRowFn.java
@@ -23,7 +23,7 @@
import com.google.cloud.bigquery.StandardSQLTypeName;
import com.google.common.collect.Maps;
import com.google.protobuf.Timestamp;
-import feast.storage.bigquery.ValueBigQueryBuilder;
+import feast.store.warehouse.bigquery.ValueBigQueryBuilder;
import java.util.Map;
import org.apache.avro.generic.GenericRecord;
import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord;
@@ -34,8 +34,6 @@
import feast.types.FeatureProto.Feature;
import feast.types.FeatureRowProto.FeatureRow;
import feast.types.ValueProto.Value;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* This is a serializable function used with the BigQueryIO for fetching feature rows directly from
diff --git a/ingestion/src/main/java/feast/source/common/ValueMapToFeatureRowTransform.java b/ingestion/src/main/java/feast/source/common/ValueMapToFeatureRowTransform.java
new file mode 100644
index 00000000000..490eb62bd61
--- /dev/null
+++ b/ingestion/src/main/java/feast/source/common/ValueMapToFeatureRowTransform.java
@@ -0,0 +1,122 @@
+/*
+ * Copyright 2019 The Feast Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package feast.source.common;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.Maps;
+import feast.ingestion.metrics.FeastMetrics;
+import feast.ingestion.model.Values;
+import feast.specs.ImportSpecProto.Field;
+import feast.specs.ImportSpecProto.Schema;
+import feast.types.FeatureProto.Feature;
+import feast.types.FeatureRowProto.FeatureRow;
+import feast.types.ValueProto.Value;
+import feast.types.ValueProto.Value.ValCase;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.beam.sdk.extensions.protobuf.ProtoCoder;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+
+public class ValueMapToFeatureRowTransform extends
+ PTransform>, PCollection> {
+
+ private String entity;
+ private Schema schema;
+
+ public ValueMapToFeatureRowTransform(String entity, Schema schema) {
+ this.entity = entity;
+ this.schema = schema;
+ }
+
+ @Override
+ public PCollection expand(PCollection> input) {
+ return input.apply(ParDo.of(ValueMapToFeatureRowDoFn.of(entity, schema)))
+ .setCoder(ProtoCoder.of(FeatureRow.class));
+ }
+
+ public static class ValueMapToFeatureRowDoFn extends
+ DoFn, FeatureRow> {
+
+ private String entityIdColumn;
+ private String timestampColumn;
+ private com.google.protobuf.Timestamp timestampValue;
+ private String entity;
+ private Map fields;
+
+ private ValueMapToFeatureRowDoFn(String entity, Map fields,
+ String entityIdColumn,
+ String timestampColumn,
+ com.google.protobuf.Timestamp timestampValue) {
+ this.entity = entity;
+ this.fields = fields;
+ this.entityIdColumn = entityIdColumn;
+ this.timestampColumn = timestampColumn;
+ this.timestampValue = timestampValue;
+ }
+
+ public static ValueMapToFeatureRowDoFn of(String entity, Schema schema) {
+ final Map fields = Maps.newHashMap();
+ for (Field field : schema.getFieldsList()) {
+ String displayName = !field.getName().isEmpty() ? field.getName() : field.getFeatureId();
+ fields.put(displayName, field);
+ }
+ return new ValueMapToFeatureRowDoFn(entity, fields, schema.getEntityIdColumn(),
+ schema.getTimestampColumn(), schema.getTimestampValue());
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ FeatureRow.Builder builder = FeatureRow.newBuilder();
+ try {
+ Map valueMap = context.element();
+ builder.setEntityName(entity);
+ for (Entry entry : valueMap.entrySet()) {
+ String name = entry.getKey();
+ Value value = entry.getValue();
+ Field field = fields.get(name);
+
+ if (value == null || value.getValCase().equals(ValCase.VAL_NOT_SET)) {
+ continue;
+ }
+
+ // A feature can only be one of these things
+ if (entityIdColumn.equals(name)) {
+ builder.setEntityKey(Values.asString(value).getStringVal());
+ } else if (timestampColumn.equals(name)) {
+ builder.setEventTimestamp(Values.asTimestamp(value).getTimestampVal());
+ } else if (!Strings.isNullOrEmpty(field.getFeatureId())) {
+ String featureId = field.getFeatureId();
+ builder.addFeatures(
+ Feature.newBuilder().setId(featureId).setValue(value));
+ }
+ // else silently ignore this column
+ }
+ if (!timestampValue.equals(com.google.protobuf.Timestamp.getDefaultInstance())) {
+ // This overrides any column event timestamp.
+ builder.setEventTimestamp(timestampValue);
+ }
+ context.output(builder.build());
+ } catch (Exception e) {
+ FeastMetrics.inc(builder.build(), "input_errors");
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/ingestion/src/main/java/feast/source/csv/CsvFileFeatureSource.java b/ingestion/src/main/java/feast/source/csv/CsvFileFeatureSource.java
index 837050fc680..0d713134f39 100644
--- a/ingestion/src/main/java/feast/source/csv/CsvFileFeatureSource.java
+++ b/ingestion/src/main/java/feast/source/csv/CsvFileFeatureSource.java
@@ -24,25 +24,20 @@
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
-import feast.ingestion.metrics.FeastMetrics;
-import feast.ingestion.model.Values;
-import feast.ingestion.util.DateUtil;
import feast.options.Options;
import feast.options.OptionsParser;
import feast.source.FeatureSource;
import feast.source.FeatureSourceFactory;
-import feast.source.csv.CsvIO.StringMap;
+import feast.source.common.ValueMapToFeatureRowTransform.ValueMapToFeatureRowDoFn;
import feast.specs.ImportSpecProto.Field;
import feast.specs.ImportSpecProto.ImportSpec;
import feast.specs.ImportSpecProto.Schema;
-import feast.types.FeatureProto.Feature;
import feast.types.FeatureRowProto.FeatureRow;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
import javax.validation.constraints.NotEmpty;
import lombok.AllArgsConstructor;
-import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PInput;
@@ -70,7 +65,7 @@ public class CsvFileFeatureSource extends FeatureSource {
@Override
public PCollection expand(PInput input) {
CsvFileFeatureSourceOptions options = OptionsParser
- .parse(importSpec.getOptionsMap(), CsvFileFeatureSourceOptions.class);
+ .parse(importSpec.getSourceOptionsMap(), CsvFileFeatureSourceOptions.class);
List entities = importSpec.getEntitiesList();
Preconditions.checkArgument(
entities.size() == 1, "exactly 1 entity must be set for CSV import");
@@ -94,50 +89,16 @@ public PCollection expand(PInput input) {
schema.getFieldsList().size() > 0,
"CSV import needs schema with a least one field specified");
- PCollection