specPublisher) {
this.featureSetRepository = featureSetRepository;
this.storeRepository = storeRepository;
this.projectRepository = projectRepository;
this.defaultSource = defaultSource;
+ this.specPublisher = specPublisher;
}
/**
@@ -240,6 +249,7 @@ public ListStoresResponse listStores(ListStoresRequest.Filter filter) {
*
* @param newFeatureSet Feature set that will be created or updated.
*/
+ @Transactional
public ApplyFeatureSetResponse applyFeatureSet(FeatureSetProto.FeatureSet newFeatureSet)
throws InvalidProtocolBufferException {
// Autofill default project if not specified
@@ -299,6 +309,39 @@ public ApplyFeatureSetResponse applyFeatureSet(FeatureSetProto.FeatureSet newFea
status = Status.UPDATED;
}
+ featureSet.incVersion();
+
+ // Sending latest version of FeatureSet to all currently running IngestionJobs (there's one
+ // topic for all sets).
+ // All related jobs would apply new FeatureSet on the fly.
+ // We wait for Kafka broker to ack that the message was added to topic before actually
+ // committing this FeatureSet.
+ // In case kafka doesn't respond within SPEC_PUBLISHING_TIMEOUT_SECONDS we abort current
+ // transaction and return error to client.
+ try {
+ specPublisher
+ .sendDefault(featureSet.getReference(), featureSet.toProto().getSpec())
+ .get(SPEC_PUBLISHING_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw io.grpc.Status.UNAVAILABLE
+ .withDescription(
+ String.format(
+ "Unable to publish FeatureSet to Kafka. Cause: %s",
+ e.getCause() != null ? e.getCause().getMessage() : "unknown"))
+ .withCause(e)
+ .asRuntimeException();
+ }
+
+ // Updating delivery status for related jobs (that are currently using this FeatureSet).
+ // We now set status to IN_PROGRESS, so listenAckFromJobs would be able to
+ // monitor delivery progress for each new version.
+ featureSet.getJobStatuses().stream()
+ .filter(s -> s.getJob().isRunning())
+ .forEach(
+ s ->
+ s.setDeliveryStatus(
+ FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_IN_PROGRESS));
+
// Persist the FeatureSet object
featureSet.setStatus(FeatureSetStatus.STATUS_PENDING);
project.addFeatureSet(featureSet);
@@ -347,4 +390,60 @@ public UpdateStoreResponse updateStore(UpdateStoreRequest updateStoreRequest)
.setStore(updateStoreRequest.getStore())
.build();
}
+
+ /**
+ * Listener for ACK messages coming from IngestionJob when FeatureSetSpec is installed (in
+ * pipeline).
+ *
+ * Updates FeatureSetJobStatus for respected FeatureSet (selected by reference) and Job (select
+ * by Id).
+ *
+ *
When all related (running) to FeatureSet jobs are updated - FeatureSet receives READY status
+ *
+ * @param record ConsumerRecord with key: FeatureSet reference and value: Ack message
+ */
+ @KafkaListener(topics = {"${feast.stream.specsOptions.specsAckTopic}"})
+ @Transactional
+ public void listenAckFromJobs(
+ ConsumerRecord record) {
+ String setReference = record.key();
+ Pair projectAndSetName = parseReference(setReference);
+ FeatureSet featureSet =
+ featureSetRepository.findFeatureSetByNameAndProject_Name(
+ projectAndSetName.getSecond(), projectAndSetName.getFirst());
+ if (featureSet == null) {
+ log.warn(
+ String.format("ACKListener received message for unknown FeatureSet %s", setReference));
+ return;
+ }
+
+ if (featureSet.getVersion() != record.value().getFeatureSetVersion()) {
+ log.warn(
+ String.format(
+ "ACKListener received outdated ack for %s. Current %d, Received %d",
+ setReference, featureSet.getVersion(), record.value().getFeatureSetVersion()));
+ return;
+ }
+
+ featureSet.getJobStatuses().stream()
+ .filter(js -> js.getJob().getId().equals(record.value().getJobName()))
+ .findFirst()
+ .ifPresent(
+ featureSetJobStatus ->
+ featureSetJobStatus.setDeliveryStatus(
+ FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_DELIVERED));
+
+ boolean allDelivered =
+ featureSet.getJobStatuses().stream()
+ .filter(js -> js.getJob().isRunning())
+ .allMatch(
+ js ->
+ js.getDeliveryStatus()
+ .equals(FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_DELIVERED));
+
+ if (allDelivered) {
+ featureSet.setStatus(FeatureSetStatus.STATUS_READY);
+ featureSetRepository.saveAndFlush(featureSet);
+ }
+ }
}
diff --git a/core/src/main/java/feast/core/util/KafkaSerialization.java b/core/src/main/java/feast/core/util/KafkaSerialization.java
new file mode 100644
index 00000000000..881d8eb9cf8
--- /dev/null
+++ b/core/src/main/java/feast/core/util/KafkaSerialization.java
@@ -0,0 +1,68 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 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.core.util;
+
+import com.google.protobuf.GeneratedMessageV3;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Message;
+import com.google.protobuf.Parser;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serializer;
+
+/*
+Serializer & Deserializer implementation to write & read protobuf object from/to kafka
+ */
+public class KafkaSerialization {
+ public static class ProtoSerializer implements Serializer {
+ @Override
+ public byte[] serialize(String topic, T data) {
+ ByteArrayOutputStream stream = new ByteArrayOutputStream();
+ try {
+ data.writeTo(stream);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ String.format(
+ "Unable to serialize object of type %s. Reason: %s",
+ data.getClass().getName(), e.getCause().getMessage()));
+ }
+
+ return stream.toByteArray();
+ }
+ }
+
+ public static class ProtoDeserializer implements Deserializer {
+ private Parser parser;
+
+ public ProtoDeserializer(Parser parser) {
+ this.parser = parser;
+ }
+
+ @Override
+ public T deserialize(String topic, byte[] data) {
+ try {
+ return parser.parseFrom(data);
+ } catch (InvalidProtocolBufferException e) {
+ throw new RuntimeException(
+ String.format(
+ "Unable to deserialize object from topic %s. Reason: %s",
+ topic, e.getCause().getMessage()));
+ }
+ }
+ }
+}
diff --git a/core/src/main/resources/application-override.yaml b/core/src/main/resources/application-override.yaml
deleted file mode 100644
index e69de29bb2d..00000000000
diff --git a/core/src/main/resources/application.yml b/core/src/main/resources/application.yml
index 5dc5698e889..9f02ed302db 100644
--- a/core/src/main/resources/application.yml
+++ b/core/src/main/resources/application.yml
@@ -78,6 +78,10 @@ feast:
replicationFactor: 1
partitions: 1
+ specsOptions:
+ specsTopic: feast-specs
+ specsAckTopic: feast-specs-ack
+
spring:
jpa:
properties.hibernate:
diff --git a/core/src/test/java/feast/core/job/JobUpdateTaskTest.java b/core/src/test/java/feast/core/job/JobUpdateTaskTest.java
index d1826738019..fc633ba15f8 100644
--- a/core/src/test/java/feast/core/job/JobUpdateTaskTest.java
+++ b/core/src/test/java/feast/core/job/JobUpdateTaskTest.java
@@ -24,11 +24,8 @@
import static org.mockito.Mockito.when;
import static org.mockito.MockitoAnnotations.initMocks;
-import feast.core.model.FeatureSet;
-import feast.core.model.Job;
-import feast.core.model.JobStatus;
-import feast.core.model.Source;
-import feast.core.model.Store;
+import feast.core.model.*;
+import feast.core.util.ModelHelpers;
import feast.proto.core.FeatureSetProto;
import feast.proto.core.FeatureSetProto.FeatureSetMeta;
import feast.proto.core.FeatureSetProto.FeatureSetSpec;
@@ -92,7 +89,14 @@ public void setUp() {
}
Job makeJob(String extId, List featureSets, JobStatus status) {
- return new Job("job", extId, RUNNER, source, store, featureSets, status);
+ return new Job(
+ "job",
+ extId,
+ RUNNER,
+ source,
+ store,
+ ModelHelpers.makeFeatureSetJobStatus(featureSets),
+ status);
}
JobUpdateTask makeTask(List featureSets, Optional currentJob) {
diff --git a/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java b/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java
index ea9caa91ff6..50812882029 100644
--- a/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java
+++ b/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java
@@ -26,20 +26,19 @@
import com.google.api.client.googleapis.testing.auth.oauth2.MockGoogleCredential;
import com.google.api.services.dataflow.Dataflow;
import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
import com.google.protobuf.Duration;
import com.google.protobuf.util.JsonFormat;
import com.google.protobuf.util.JsonFormat.Printer;
import feast.core.config.FeastProperties.MetricsProperties;
import feast.core.exception.JobExecutionException;
import feast.core.job.Runner;
-import feast.core.job.option.FeatureSetJsonByteConverter;
import feast.core.model.*;
-import feast.ingestion.options.BZip2Compressor;
import feast.ingestion.options.ImportOptions;
-import feast.ingestion.options.OptionCompressor;
import feast.proto.core.FeatureSetProto;
import feast.proto.core.FeatureSetProto.FeatureSetMeta;
import feast.proto.core.FeatureSetProto.FeatureSetSpec;
+import feast.proto.core.IngestionJobProto;
import feast.proto.core.RunnerProto.DataflowRunnerConfigOptions;
import feast.proto.core.RunnerProto.DataflowRunnerConfigOptions.Builder;
import feast.proto.core.SourceProto;
@@ -50,8 +49,6 @@
import feast.proto.core.StoreProto.Store.StoreType;
import feast.proto.core.StoreProto.Store.Subscription;
import java.io.IOException;
-import java.util.Collections;
-import java.util.List;
import org.apache.beam.runners.dataflow.DataflowPipelineJob;
import org.apache.beam.runners.dataflow.DataflowRunner;
import org.apache.beam.sdk.PipelineResult.State;
@@ -71,6 +68,7 @@ public class DataflowJobManagerTest {
@Mock private Dataflow dataflow;
private DataflowRunnerConfigOptions defaults;
+ private IngestionJobProto.SpecsStreamingUpdateConfig specsStreamingUpdateConfig;
private DataflowJobManager dfJobManager;
@Before
@@ -94,7 +92,17 @@ public void setUp() {
e.printStackTrace();
}
- dfJobManager = new DataflowJobManager(defaults, metricsProperties, credential);
+ specsStreamingUpdateConfig =
+ IngestionJobProto.SpecsStreamingUpdateConfig.newBuilder()
+ .setSource(
+ KafkaSourceConfig.newBuilder()
+ .setTopic("specs_topic")
+ .setBootstrapServers("servers:9092")
+ .build())
+ .build();
+
+ dfJobManager =
+ new DataflowJobManager(defaults, metricsProperties, specsStreamingUpdateConfig, credential);
dfJobManager = spy(dfJobManager);
}
@@ -142,11 +150,7 @@ public void shouldStartJobWithCorrectPipelineOptions() throws IOException {
expectedPipelineOptions.setLabels(defaults.getLabelsMap());
expectedPipelineOptions.setJobName(jobName);
expectedPipelineOptions.setStoreJson(Lists.newArrayList(printer.print(store)));
-
- OptionCompressor> featureSetJsonCompressor =
- new BZip2Compressor<>(new FeatureSetJsonByteConverter());
- expectedPipelineOptions.setFeatureSetJson(
- featureSetJsonCompressor.compress(Collections.singletonList(featureSet)));
+ expectedPipelineOptions.setSourceJson(printer.print(source));
ArgumentCaptor captor = ArgumentCaptor.forClass(ImportOptions.class);
@@ -155,6 +159,10 @@ public void shouldStartJobWithCorrectPipelineOptions() throws IOException {
when(mockPipelineResult.getJobId()).thenReturn(expectedExtJobId);
doReturn(mockPipelineResult).when(dfJobManager).runPipeline(any());
+
+ FeatureSetJobStatus featureSetJobStatus = new FeatureSetJobStatus();
+ featureSetJobStatus.setFeatureSet(FeatureSet.fromProto(featureSet));
+
Job job =
new Job(
jobName,
@@ -162,7 +170,7 @@ public void shouldStartJobWithCorrectPipelineOptions() throws IOException {
Runner.DATAFLOW,
Source.fromProto(source),
Store.fromProto(store),
- Lists.newArrayList(FeatureSet.fromProto(featureSet)),
+ Sets.newHashSet(featureSetJobStatus),
JobStatus.PENDING);
Job actual = dfJobManager.startJob(job);
@@ -184,9 +192,6 @@ public void shouldStartJobWithCorrectPipelineOptions() throws IOException {
// Assume the files that are staged are correct
expectedPipelineOptions.setFilesToStage(actualPipelineOptions.getFilesToStage());
- assertThat(
- actualPipelineOptions.getFeatureSetJson(),
- equalTo(expectedPipelineOptions.getFeatureSetJson()));
assertThat(
actualPipelineOptions.getDeadLetterTableSpec(),
equalTo(expectedPipelineOptions.getDeadLetterTableSpec()));
@@ -197,6 +202,11 @@ public void shouldStartJobWithCorrectPipelineOptions() throws IOException {
equalTo(expectedPipelineOptions.getMetricsExporterType()));
assertThat(
actualPipelineOptions.getStoreJson(), equalTo(expectedPipelineOptions.getStoreJson()));
+ assertThat(
+ actualPipelineOptions.getSourceJson(), equalTo(expectedPipelineOptions.getSourceJson()));
+ assertThat(
+ actualPipelineOptions.getSpecsStreamingUpdateConfigJson(),
+ equalTo(printer.print(specsStreamingUpdateConfig)));
assertThat(actual.getExtId(), equalTo(expectedExtJobId));
}
@@ -231,6 +241,9 @@ public void shouldThrowExceptionWhenJobStateTerminal() throws IOException {
doReturn(mockPipelineResult).when(dfJobManager).runPipeline(any());
+ FeatureSetJobStatus featureSetJobStatus = new FeatureSetJobStatus();
+ featureSetJobStatus.setFeatureSet(FeatureSet.fromProto(featureSet));
+
Job job =
new Job(
"job",
@@ -238,7 +251,7 @@ public void shouldThrowExceptionWhenJobStateTerminal() throws IOException {
Runner.DATAFLOW,
Source.fromProto(source),
Store.fromProto(store),
- Lists.newArrayList(FeatureSet.fromProto(featureSet)),
+ Sets.newHashSet(featureSetJobStatus),
JobStatus.PENDING);
expectedException.expect(JobExecutionException.class);
diff --git a/core/src/test/java/feast/core/job/direct/DirectRunnerJobManagerTest.java b/core/src/test/java/feast/core/job/direct/DirectRunnerJobManagerTest.java
index 0128f5aa0b3..0e797dc64d4 100644
--- a/core/src/test/java/feast/core/job/direct/DirectRunnerJobManagerTest.java
+++ b/core/src/test/java/feast/core/job/direct/DirectRunnerJobManagerTest.java
@@ -16,6 +16,7 @@
*/
package feast.core.job.direct;
+import static feast.core.util.ModelHelpers.makeFeatureSetJobStatus;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;
import static org.mockito.ArgumentMatchers.any;
@@ -31,17 +32,15 @@
import com.google.protobuf.util.JsonFormat.Printer;
import feast.core.config.FeastProperties.MetricsProperties;
import feast.core.job.Runner;
-import feast.core.job.option.FeatureSetJsonByteConverter;
import feast.core.model.FeatureSet;
import feast.core.model.Job;
import feast.core.model.JobStatus;
import feast.core.model.Source;
import feast.core.model.Store;
-import feast.ingestion.options.BZip2Compressor;
import feast.ingestion.options.ImportOptions;
-import feast.ingestion.options.OptionCompressor;
import feast.proto.core.FeatureSetProto;
import feast.proto.core.FeatureSetProto.FeatureSetSpec;
+import feast.proto.core.IngestionJobProto;
import feast.proto.core.RunnerProto.DirectRunnerConfigOptions;
import feast.proto.core.SourceProto;
import feast.proto.core.SourceProto.KafkaSourceConfig;
@@ -51,8 +50,6 @@
import feast.proto.core.StoreProto.Store.StoreType;
import feast.proto.core.StoreProto.Store.Subscription;
import java.io.IOException;
-import java.util.Collections;
-import java.util.List;
import org.apache.beam.runners.direct.DirectRunner;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
@@ -71,6 +68,7 @@ public class DirectRunnerJobManagerTest {
private DirectRunnerJobManager drJobManager;
private DirectRunnerConfigOptions defaults;
+ private IngestionJobProto.SpecsStreamingUpdateConfig specsStreamingUpdateConfig;
@Before
public void setUp() {
@@ -79,7 +77,18 @@ public void setUp() {
MetricsProperties metricsProperties = new MetricsProperties();
metricsProperties.setEnabled(false);
- drJobManager = new DirectRunnerJobManager(defaults, directJobRegistry, metricsProperties);
+ specsStreamingUpdateConfig =
+ IngestionJobProto.SpecsStreamingUpdateConfig.newBuilder()
+ .setSource(
+ KafkaSourceConfig.newBuilder()
+ .setTopic("specs_topic")
+ .setBootstrapServers("servers:9092")
+ .build())
+ .build();
+
+ drJobManager =
+ new DirectRunnerJobManager(
+ defaults, directJobRegistry, metricsProperties, specsStreamingUpdateConfig);
drJobManager = Mockito.spy(drJobManager);
}
@@ -125,11 +134,7 @@ public void shouldStartDirectJobAndRegisterPipelineResult() throws IOException {
expectedPipelineOptions.setTargetParallelism(1);
expectedPipelineOptions.setStoreJson(Lists.newArrayList(printer.print(store)));
expectedPipelineOptions.setProject("");
-
- OptionCompressor> featureSetJsonCompressor =
- new BZip2Compressor<>(new FeatureSetJsonByteConverter());
- expectedPipelineOptions.setFeatureSetJson(
- featureSetJsonCompressor.compress(Collections.singletonList(featureSet)));
+ expectedPipelineOptions.setSourceJson(printer.print(source));
ArgumentCaptor pipelineOptionsCaptor =
ArgumentCaptor.forClass(ImportOptions.class);
@@ -145,7 +150,7 @@ public void shouldStartDirectJobAndRegisterPipelineResult() throws IOException {
Runner.DIRECT,
Source.fromProto(source),
Store.fromProto(store),
- Lists.newArrayList(FeatureSet.fromProto(featureSet)),
+ makeFeatureSetJobStatus(FeatureSet.fromProto(featureSet)),
JobStatus.PENDING);
Job actual = drJobManager.startJob(job);
verify(drJobManager, times(1)).runPipeline(pipelineOptionsCaptor.capture());
@@ -156,9 +161,6 @@ public void shouldStartDirectJobAndRegisterPipelineResult() throws IOException {
expectedPipelineOptions.setOptionsId(
actualPipelineOptions.getOptionsId()); // avoid comparing this value
- assertThat(
- actualPipelineOptions.getFeatureSetJson(),
- equalTo(expectedPipelineOptions.getFeatureSetJson()));
assertThat(
actualPipelineOptions.getDeadLetterTableSpec(),
equalTo(expectedPipelineOptions.getDeadLetterTableSpec()));
@@ -169,6 +171,11 @@ public void shouldStartDirectJobAndRegisterPipelineResult() throws IOException {
equalTo(expectedPipelineOptions.getMetricsExporterType()));
assertThat(
actualPipelineOptions.getStoreJson(), equalTo(expectedPipelineOptions.getStoreJson()));
+ assertThat(
+ actualPipelineOptions.getSourceJson(), equalTo(expectedPipelineOptions.getSourceJson()));
+ assertThat(
+ actualPipelineOptions.getSpecsStreamingUpdateConfigJson(),
+ equalTo(printer.print(specsStreamingUpdateConfig)));
assertThat(jobStarted.getPipelineResult(), equalTo(mockPipelineResult));
assertThat(jobStarted.getJobId(), equalTo(expectedJobId));
diff --git a/core/src/test/java/feast/core/service/JobCoordinatorServiceTest.java b/core/src/test/java/feast/core/service/JobCoordinatorServiceTest.java
index 8386efb28f4..caec99f2287 100644
--- a/core/src/test/java/feast/core/service/JobCoordinatorServiceTest.java
+++ b/core/src/test/java/feast/core/service/JobCoordinatorServiceTest.java
@@ -34,9 +34,8 @@
import feast.core.job.JobManager;
import feast.core.job.JobMatcher;
import feast.core.job.Runner;
-import feast.core.model.FeatureSet;
-import feast.core.model.Job;
-import feast.core.model.JobStatus;
+import feast.core.model.*;
+import feast.core.util.ModelHelpers;
import feast.proto.core.CoreServiceProto.ListFeatureSetsRequest.Filter;
import feast.proto.core.CoreServiceProto.ListFeatureSetsResponse;
import feast.proto.core.CoreServiceProto.ListStoresResponse;
@@ -50,7 +49,6 @@
import feast.proto.core.StoreProto.Store.RedisConfig;
import feast.proto.core.StoreProto.Store.StoreType;
import feast.proto.core.StoreProto.Store.Subscription;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.junit.Before;
@@ -159,7 +157,7 @@ public void shouldGenerateAndSubmitJobsIfAny() throws InvalidProtocolBufferExcep
Runner.DATAFLOW,
feast.core.model.Source.fromProto(source),
feast.core.model.Store.fromProto(store),
- Arrays.asList(featureSet1, featureSet2),
+ ModelHelpers.makeFeatureSetJobStatus(featureSet1, featureSet2),
JobStatus.PENDING);
Job expected =
@@ -169,7 +167,7 @@ public void shouldGenerateAndSubmitJobsIfAny() throws InvalidProtocolBufferExcep
Runner.DATAFLOW,
feast.core.model.Source.fromProto(source),
feast.core.model.Store.fromProto(store),
- Arrays.asList(featureSet1, featureSet2),
+ ModelHelpers.makeFeatureSetJobStatus(featureSet1, featureSet2),
JobStatus.RUNNING);
when(featureSetRepository.findAllByNameLikeAndProject_NameLikeOrderByNameAsc("%", "project1"))
@@ -246,7 +244,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
Runner.DATAFLOW,
feast.core.model.Source.fromProto(source1),
feast.core.model.Store.fromProto(store),
- Arrays.asList(featureSet1),
+ ModelHelpers.makeFeatureSetJobStatus(featureSet1),
JobStatus.PENDING);
Job expected1 =
@@ -256,7 +254,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
Runner.DATAFLOW,
feast.core.model.Source.fromProto(source1),
feast.core.model.Store.fromProto(store),
- Arrays.asList(featureSet1),
+ ModelHelpers.makeFeatureSetJobStatus(featureSet1),
JobStatus.RUNNING);
Job expectedInput2 =
@@ -266,7 +264,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
Runner.DATAFLOW,
feast.core.model.Source.fromProto(source2),
feast.core.model.Store.fromProto(store),
- Arrays.asList(featureSet2),
+ ModelHelpers.makeFeatureSetJobStatus(featureSet2),
JobStatus.PENDING);
Job expected2 =
@@ -276,7 +274,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
Runner.DATAFLOW,
feast.core.model.Source.fromProto(source2),
feast.core.model.Store.fromProto(store),
- Arrays.asList(featureSet2),
+ ModelHelpers.makeFeatureSetJobStatus(featureSet2),
JobStatus.RUNNING);
ArgumentCaptor> jobArgCaptor = ArgumentCaptor.forClass(List.class);
diff --git a/core/src/test/java/feast/core/service/JobServiceTest.java b/core/src/test/java/feast/core/service/JobServiceTest.java
index ff056287f9b..27509620ff2 100644
--- a/core/src/test/java/feast/core/service/JobServiceTest.java
+++ b/core/src/test/java/feast/core/service/JobServiceTest.java
@@ -16,6 +16,7 @@
*/
package feast.core.service;
+import static feast.core.util.ModelHelpers.makeFeatureSetJobStatus;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.fail;
@@ -25,6 +26,7 @@
import static org.mockito.Mockito.when;
import static org.mockito.MockitoAnnotations.initMocks;
+import com.google.common.collect.Lists;
import com.google.protobuf.InvalidProtocolBufferException;
import feast.core.dao.JobRepository;
import feast.core.job.JobManager;
@@ -126,7 +128,8 @@ public void setupJobRepository() {
when(this.jobRepository.findById(this.job.getId())).thenReturn(Optional.of(this.job));
when(this.jobRepository.findByStoreName(this.dataStore.getName()))
.thenReturn(Arrays.asList(this.job));
- when(this.jobRepository.findByFeatureSetsIn(Arrays.asList(this.featureSet)))
+ when(this.jobRepository.findByFeatureSetJobStatusesIn(
+ Lists.newArrayList((this.featureSet.getJobStatuses()))))
.thenReturn(Arrays.asList(this.job));
when(this.jobRepository.findAll()).thenReturn(Arrays.asList(this.job));
}
@@ -155,7 +158,7 @@ private Job newDummyJob(String id, String extId, JobStatus status) {
Runner.DATAFLOW,
this.dataSource,
this.dataStore,
- Arrays.asList(this.featureSet),
+ makeFeatureSetJobStatus(this.featureSet),
status);
}
diff --git a/core/src/test/java/feast/core/service/SpecServiceTest.java b/core/src/test/java/feast/core/service/SpecServiceTest.java
index b5fd03fc7fc..9ec219b6c97 100644
--- a/core/src/test/java/feast/core/service/SpecServiceTest.java
+++ b/core/src/test/java/feast/core/service/SpecServiceTest.java
@@ -16,12 +16,13 @@
*/
package feast.core.service;
+import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.core.IsEqual.equalTo;
import static org.junit.Assert.assertEquals;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.*;
import static org.mockito.MockitoAnnotations.initMocks;
import com.google.api.client.util.Lists;
@@ -45,16 +46,21 @@
import feast.proto.core.FeatureSetProto.EntitySpec;
import feast.proto.core.FeatureSetProto.FeatureSetSpec;
import feast.proto.core.FeatureSetProto.FeatureSpec;
+import feast.proto.core.IngestionJobProto;
import feast.proto.core.StoreProto;
import feast.proto.core.StoreProto.Store.RedisConfig;
import feast.proto.core.StoreProto.Store.StoreType;
import feast.proto.core.StoreProto.Store.Subscription;
import feast.proto.types.ValueProto.ValueType.Enum;
+import io.grpc.StatusRuntimeException;
import java.sql.Date;
import java.time.Instant;
import java.util.*;
import java.util.Map.Entry;
+import java.util.concurrent.CancellationException;
import java.util.stream.Collectors;
+import lombok.SneakyThrows;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -62,6 +68,8 @@
import org.mockito.ArgumentCaptor;
import org.mockito.ArgumentMatchers;
import org.mockito.Mock;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.scheduling.annotation.AsyncResult;
import org.tensorflow.metadata.v0.BoolDomain;
import org.tensorflow.metadata.v0.FeaturePresence;
import org.tensorflow.metadata.v0.FeaturePresenceWithinGroup;
@@ -86,6 +94,8 @@ public class SpecServiceTest {
@Mock private ProjectRepository projectRepository;
+ @Mock private KafkaTemplate kafkaTemplate;
+
@Rule public final ExpectedException expectedException = ExpectedException.none();
private SpecService specService;
@@ -143,8 +153,11 @@ public void setUp() {
when(storeRepository.findById("SERVING")).thenReturn(Optional.of(store1));
when(storeRepository.findById("NOTFOUND")).thenReturn(Optional.empty());
+ when(kafkaTemplate.sendDefault(any(), any())).thenReturn(new AsyncResult<>(null));
+
specService =
- new SpecService(featureSetRepository, storeRepository, projectRepository, defaultSource);
+ new SpecService(
+ featureSetRepository, storeRepository, projectRepository, defaultSource, kafkaTemplate);
}
@Test
@@ -284,6 +297,7 @@ public void applyFeatureSetShouldApplyFeatureSetIfNotExists()
.build();
assertThat(applyFeatureSetResponse.getStatus(), equalTo(Status.CREATED));
assertThat(applyFeatureSetResponse.getFeatureSet().getSpec(), equalTo(expected.getSpec()));
+ assertThat(applyFeatureSetResponse.getFeatureSet().getSpec().getVersion(), equalTo(1));
}
@Test
@@ -308,7 +322,12 @@ public void applyFeatureSetShouldUpdateAndSaveFeatureSetIfAlreadyExists()
.toBuilder()
.setMeta(incomingFeatureSet.getMeta().toBuilder().build())
.setSpec(
- incomingFeatureSet.getSpec().toBuilder().setSource(defaultSource.toProto()).build())
+ incomingFeatureSet
+ .getSpec()
+ .toBuilder()
+ .setVersion(2)
+ .setSource(defaultSource.toProto())
+ .build())
.build();
ApplyFeatureSetResponse applyFeatureSetResponse =
@@ -318,6 +337,74 @@ public void applyFeatureSetShouldUpdateAndSaveFeatureSetIfAlreadyExists()
assertEquals(
FeatureSet.fromProto(applyFeatureSetResponse.getFeatureSet()),
FeatureSet.fromProto(expected));
+
+ assertThat(applyFeatureSetResponse.getFeatureSet().getSpec().getVersion(), equalTo(2));
+ verify(kafkaTemplate)
+ .sendDefault(eq(featureSets.get(0).getReference()), any(FeatureSetSpec.class));
+ }
+
+ @Test
+ @SneakyThrows
+ public void applyFeatureSetShouldNotWorkWithoutKafkaAck() {
+ FeatureSet fsInTest = featureSets.get(1);
+ FeatureSetProto.FeatureSet incomingFeatureSet = fsInTest.toProto();
+ CancellationException exc = new CancellationException();
+ when(kafkaTemplate.sendDefault(eq(fsInTest.getReference()), any()).get()).thenThrow(exc);
+
+ incomingFeatureSet =
+ incomingFeatureSet
+ .toBuilder()
+ .setMeta(incomingFeatureSet.getMeta())
+ .setSpec(
+ incomingFeatureSet
+ .getSpec()
+ .toBuilder()
+ .addFeatures(
+ FeatureSpec.newBuilder().setName("feature2").setValueType(Enum.STRING))
+ .build())
+ .build();
+
+ expectedException.expect(StatusRuntimeException.class);
+ specService.applyFeatureSet(incomingFeatureSet);
+ verify(featureSetRepository, never()).saveAndFlush(ArgumentMatchers.any(FeatureSet.class));
+ }
+
+ @Test
+ @SneakyThrows
+ public void applyFeatureSetShouldUpdateDeliveryStatuses() {
+ FeatureSet fsInTest = featureSets.get(1);
+ FeatureSetJobStatus j1 =
+ newJob(
+ fsInTest,
+ JobStatus.RUNNING,
+ FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_DELIVERED);
+ FeatureSetJobStatus j2 =
+ newJob(
+ fsInTest,
+ JobStatus.ABORTED,
+ FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_DELIVERED);
+
+ fsInTest.getJobStatuses().addAll(Arrays.asList(j1, j2));
+
+ FeatureSetProto.FeatureSet incomingFeatureSet = fsInTest.toProto();
+ incomingFeatureSet =
+ incomingFeatureSet
+ .toBuilder()
+ .setMeta(incomingFeatureSet.getMeta())
+ .setSpec(
+ incomingFeatureSet
+ .getSpec()
+ .toBuilder()
+ .addFeatures(
+ FeatureSpec.newBuilder().setName("feature2").setValueType(Enum.STRING))
+ .build())
+ .build();
+
+ specService.applyFeatureSet(incomingFeatureSet);
+ assertThat(
+ j1.getDeliveryStatus(), is(FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_IN_PROGRESS));
+ assertThat(
+ j2.getDeliveryStatus(), is(FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_DELIVERED));
}
@Test
@@ -681,6 +768,70 @@ public void getOrListFeatureSetShouldUseDefaultProjectIfProjectUnspecified()
assertThat(listResponse.getFeatureSetsList(), equalTo(Arrays.asList(expected.toProto())));
}
+ @Test
+ public void specAckListenerShouldDoNothingWhenMessageIsOutdated() {
+ FeatureSet fsInTest = featureSets.get(1);
+ FeatureSetJobStatus j1 =
+ newJob(
+ fsInTest,
+ JobStatus.RUNNING,
+ FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_IN_PROGRESS);
+ FeatureSetJobStatus j2 =
+ newJob(
+ fsInTest,
+ JobStatus.RUNNING,
+ FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_IN_PROGRESS);
+
+ fsInTest.getJobStatuses().addAll(Arrays.asList(j1, j2));
+
+ specService.listenAckFromJobs(newAckMessage("project/invalid", 0, j1.getJob().getId()));
+ specService.listenAckFromJobs(newAckMessage(fsInTest.getReference(), 0, ""));
+ specService.listenAckFromJobs(newAckMessage(fsInTest.getReference(), -1, j1.getJob().getId()));
+
+ assertThat(
+ j1.getDeliveryStatus(), is(FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_IN_PROGRESS));
+ assertThat(
+ j2.getDeliveryStatus(), is(FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_IN_PROGRESS));
+ }
+
+ @Test
+ public void specAckListenerShouldUpdateFeatureSetStatus() {
+ FeatureSet fsInTest = featureSets.get(1);
+ fsInTest.setStatus(FeatureSetProto.FeatureSetStatus.STATUS_PENDING);
+
+ FeatureSetJobStatus j1 =
+ newJob(
+ fsInTest,
+ JobStatus.RUNNING,
+ FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_IN_PROGRESS);
+ FeatureSetJobStatus j2 =
+ newJob(
+ fsInTest,
+ JobStatus.RUNNING,
+ FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_IN_PROGRESS);
+ FeatureSetJobStatus j3 =
+ newJob(
+ fsInTest,
+ JobStatus.ABORTED,
+ FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_IN_PROGRESS);
+
+ fsInTest.getJobStatuses().addAll(Arrays.asList(j1, j2, j3));
+
+ specService.listenAckFromJobs(
+ newAckMessage(fsInTest.getReference(), fsInTest.getVersion(), j1.getJob().getId()));
+
+ assertThat(
+ j1.getDeliveryStatus(), is(FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_DELIVERED));
+ assertThat(fsInTest.getStatus(), is(FeatureSetProto.FeatureSetStatus.STATUS_PENDING));
+
+ specService.listenAckFromJobs(
+ newAckMessage(fsInTest.getReference(), fsInTest.getVersion(), j2.getJob().getId()));
+
+ assertThat(
+ j2.getDeliveryStatus(), is(FeatureSetProto.FeatureSetJobDeliveryStatus.STATUS_DELIVERED));
+ assertThat(fsInTest.getStatus(), is(FeatureSetProto.FeatureSetStatus.STATUS_READY));
+ }
+
private FeatureSet newDummyFeatureSet(String name, String project) {
FeatureSpec f1 =
FeatureSpec.newBuilder()
@@ -698,6 +849,33 @@ private FeatureSet newDummyFeatureSet(String name, String project) {
return fs;
}
+ private FeatureSetJobStatus newJob(
+ FeatureSet fs, JobStatus status, FeatureSetProto.FeatureSetJobDeliveryStatus deliveryStatus) {
+ Job job = new Job();
+ job.setStatus(status);
+ job.setId(UUID.randomUUID().toString());
+
+ FeatureSetJobStatus featureSetJobStatus = new FeatureSetJobStatus();
+ featureSetJobStatus.setJob(job);
+ featureSetJobStatus.setFeatureSet(fs);
+ featureSetJobStatus.setDeliveryStatus(deliveryStatus);
+
+ return featureSetJobStatus;
+ }
+
+ private ConsumerRecord newAckMessage(
+ String key, int version, String jobName) {
+ return new ConsumerRecord<>(
+ "topic",
+ 0,
+ 0,
+ key,
+ IngestionJobProto.FeatureSetSpecAck.newBuilder()
+ .setFeatureSetVersion(version)
+ .setJobName(jobName)
+ .build());
+ }
+
private Store newDummyStore(String name) {
// Add type to this method when we enable filtering by type
Store store = new Store();
diff --git a/core/src/test/java/feast/core/service/TestObjectFactory.java b/core/src/test/java/feast/core/service/TestObjectFactory.java
index 40c379d3cf7..a498d6c21c2 100644
--- a/core/src/test/java/feast/core/service/TestObjectFactory.java
+++ b/core/src/test/java/feast/core/service/TestObjectFactory.java
@@ -39,15 +39,18 @@ public class TestObjectFactory {
public static FeatureSet CreateFeatureSet(
String name, String project, List entities, List features) {
- return new FeatureSet(
- name,
- project,
- 100L,
- entities,
- features,
- defaultSource,
- new HashMap<>(),
- FeatureSetProto.FeatureSetStatus.STATUS_READY);
+ FeatureSet fs =
+ new FeatureSet(
+ name,
+ project,
+ 100L,
+ entities,
+ features,
+ defaultSource,
+ new HashMap<>(),
+ FeatureSetProto.FeatureSetStatus.STATUS_READY);
+ fs.setVersion(1);
+ return fs;
}
public static Feature CreateFeature(String name, ValueProto.ValueType.Enum valueType) {
diff --git a/core/src/test/java/feast/core/util/ModelHelpers.java b/core/src/test/java/feast/core/util/ModelHelpers.java
new file mode 100644
index 00000000000..f864a5562de
--- /dev/null
+++ b/core/src/test/java/feast/core/util/ModelHelpers.java
@@ -0,0 +1,41 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 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.core.util;
+
+import feast.core.model.FeatureSet;
+import feast.core.model.FeatureSetJobStatus;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class ModelHelpers {
+ public static Set makeFeatureSetJobStatus(FeatureSet... featureSets) {
+ return Stream.of(featureSets)
+ .map(
+ fs -> {
+ FeatureSetJobStatus s = new FeatureSetJobStatus();
+ s.setFeatureSet(fs);
+ return s;
+ })
+ .collect(Collectors.toSet());
+ }
+
+ public static Set makeFeatureSetJobStatus(List featureSets) {
+ return makeFeatureSetJobStatus(featureSets.toArray(FeatureSet[]::new));
+ }
+}
diff --git a/infra/scripts/test-end-to-end-batch-dataflow.sh b/infra/scripts/test-end-to-end-batch-dataflow.sh
index 9ad29185570..e138a75bdea 100755
--- a/infra/scripts/test-end-to-end-batch-dataflow.sh
+++ b/infra/scripts/test-end-to-end-batch-dataflow.sh
@@ -14,6 +14,8 @@ test -z ${K8_CLUSTER_NAME} && K8_CLUSTER_NAME="feast-e2e-dataflow"
test -z ${HELM_RELEASE_NAME} && HELM_RELEASE_NAME="pr-$PULL_NUMBER"
test -z ${HELM_COMMON_NAME} && HELM_COMMON_NAME="deps"
test -z ${DATASET_NAME} && DATASET_NAME=feast_e2e_$(date +%s)
+test -z ${SPECS_TOPIC} && SPECS_TOPIC=feast-specs-$(date +%s)
+
feast_kafka_1_ip_name="feast-kafka-1"
feast_kafka_2_ip_name="feast-kafka-2"
@@ -209,8 +211,9 @@ export GCLOUD_SUBNET=$GCLOUD_SUBNET
export GCLOUD_REGION=$GCLOUD_REGION
export HELM_COMMON_NAME=$HELM_COMMON_NAME
export IMAGE_TAG=${PULL_PULL_SHA:1}
+export SPECS_TOPIC=$SPECS_TOPIC
-envsubst $'$TEMP_BUCKET $DATASET_NAME $GCLOUD_PROJECT $GCLOUD_NETWORK \
+envsubst $'$TEMP_BUCKET $DATASET_NAME $GCLOUD_PROJECT $GCLOUD_NETWORK $SPECS_TOPIC \
$GCLOUD_SUBNET $GCLOUD_REGION $IMAGE_TAG $HELM_COMMON_NAME $feast_kafka_1_ip
$feast_kafka_2_ip $feast_kafka_3_ip $feast_redis_ip $feast_statsd_ip' < $ORIGINAL_DIR/infra/scripts/test-templates/values-end-to-end-batch-dataflow.yaml > $ORIGINAL_DIR/infra/charts/feast/values-end-to-end-batch-dataflow-updated.yaml
diff --git a/infra/scripts/test-templates/values-end-to-end-batch-dataflow.yaml b/infra/scripts/test-templates/values-end-to-end-batch-dataflow.yaml
index 345d5234184..48231face69 100644
--- a/infra/scripts/test-templates/values-end-to-end-batch-dataflow.yaml
+++ b/infra/scripts/test-templates/values-end-to-end-batch-dataflow.yaml
@@ -15,6 +15,9 @@ feast-core:
stream:
options:
bootstrapServers: $feast_kafka_1_ip:31090
+ specsOptions:
+ specsTopic: $SPECS_TOPIC
+ specsAckTopic: $SPECS_TOPIC-ack
jobs:
active_runner: dataflow
diff --git a/ingestion/pom.xml b/ingestion/pom.xml
index 36d5cd6269b..3eadcdca252 100644
--- a/ingestion/pom.xml
+++ b/ingestion/pom.xml
@@ -230,6 +230,13 @@
test
+
+ org.slf4j
+ slf4j-simple
+ 1.7.30
+ test
+
+
com.google.guava
guava
diff --git a/ingestion/src/main/java/feast/ingestion/ImportJob.java b/ingestion/src/main/java/feast/ingestion/ImportJob.java
index 5cef8e57449..8dee52e08a2 100644
--- a/ingestion/src/main/java/feast/ingestion/ImportJob.java
+++ b/ingestion/src/main/java/feast/ingestion/ImportJob.java
@@ -16,20 +16,20 @@
*/
package feast.ingestion;
-import static feast.ingestion.utils.SpecUtil.getFeatureSetReference;
import static feast.ingestion.utils.StoreUtil.getFeatureSink;
import com.google.protobuf.InvalidProtocolBufferException;
-import feast.ingestion.options.BZip2Decompressor;
import feast.ingestion.options.ImportOptions;
-import feast.ingestion.options.StringListStreamConverter;
+import feast.ingestion.transform.FeatureRowToStoreAllocator;
import feast.ingestion.transform.ProcessAndValidateFeatureRows;
import feast.ingestion.transform.ReadFromSource;
import feast.ingestion.transform.metrics.WriteFailureMetricsTransform;
import feast.ingestion.transform.metrics.WriteSuccessMetricsTransform;
+import feast.ingestion.transform.specs.ReadFeatureSetSpecs;
+import feast.ingestion.transform.specs.WriteFeatureSetSpecAck;
import feast.ingestion.utils.SpecUtil;
-import feast.proto.core.FeatureSetProto.FeatureSet;
import feast.proto.core.FeatureSetProto.FeatureSetSpec;
+import feast.proto.core.IngestionJobProto.SpecsStreamingUpdateConfig;
import feast.proto.core.SourceProto.Source;
import feast.proto.core.StoreProto.Store;
import feast.proto.types.FeatureRowProto.FeatureRow;
@@ -39,15 +39,16 @@
import feast.storage.api.writer.WriteResult;
import feast.storage.connectors.bigquery.writer.BigQueryDeadletterSink;
import java.io.IOException;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.PipelineOptionsValidator;
import org.apache.beam.sdk.transforms.*;
import org.apache.beam.sdk.values.*;
+import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
public class ImportJob {
@@ -73,11 +74,14 @@ public static void main(String[] args) throws IOException {
public static PipelineResult runPipeline(ImportOptions options) throws IOException {
/*
* Steps:
- * 1. Read messages from Feast Source as FeatureRow
- * 2. Validate the feature rows to ensure the schema matches what is registered to the system
- * 3. Write FeatureRow to the corresponding Store
- * 4. Write elements that failed to be processed to a dead letter queue.
- * 5. Write metrics to a metrics sink
+ * 1. Read FeatureSetSpec messages from kafka
+ * 2. Read messages from Feast Source as FeatureRow
+ * 3. Validate the feature rows to ensure the schema matches what is registered to the system
+ * 4. Distribute rows across stores by subscription
+ * 5. Write FeatureRow to the corresponding Store
+ * 6. Write elements that failed to be processed to a dead letter queue.
+ * 7. Write metrics to a metrics sink
+ * 8. Send ack on receiving FeatureSetSpec
*/
PipelineOptionsValidator.validate(ImportOptions.class, options);
@@ -85,75 +89,71 @@ public static PipelineResult runPipeline(ImportOptions options) throws IOExcepti
log.info("Starting import job with settings: \n{}", options.toString());
- BZip2Decompressor> decompressor =
- new BZip2Decompressor<>(new StringListStreamConverter());
- List featureSetJson = decompressor.decompress(options.getFeatureSetJson());
- List featureSets = SpecUtil.parseFeatureSetSpecJsonList(featureSetJson);
List stores = SpecUtil.parseStoreJsonList(options.getStoreJson());
+ Source source = SpecUtil.parseSourceJson(options.getSourceJson());
+ SpecsStreamingUpdateConfig specsStreamingUpdateConfig =
+ SpecUtil.parseSpecsStreamingUpdateConfig(options.getSpecsStreamingUpdateConfigJson());
+
+ // Step 1. Read FeatureSetSpecs from Spec source
+ PCollection> featureSetSpecs =
+ pipeline.apply(
+ "ReadFeatureSetSpecs",
+ ReadFeatureSetSpecs.newBuilder()
+ .setSource(source)
+ .setStores(stores)
+ .setSpecsStreamingUpdateConfig(specsStreamingUpdateConfig)
+ .build());
+
+ PCollectionView>> globalSpecView =
+ featureSetSpecs.apply("GlobalSpecView", View.asMultimap());
+
+ // Step 2. Read messages from Feast Source as FeatureRow.
+ PCollectionTuple convertedFeatureRows =
+ pipeline.apply(
+ "ReadFeatureRowFromSource",
+ ReadFromSource.newBuilder()
+ .setSource(source)
+ .setSuccessTag(FEATURE_ROW_OUT)
+ .setFailureTag(DEADLETTER_OUT)
+ .build());
+
+ // Step 3. Process and validate incoming FeatureRows
+ PCollectionTuple validatedRows =
+ convertedFeatureRows
+ .get(FEATURE_ROW_OUT)
+ .apply(
+ ProcessAndValidateFeatureRows.newBuilder()
+ .setDefaultProject(options.getDefaultFeastProject())
+ .setFeatureSetSpecs(globalSpecView)
+ .setSuccessTag(FEATURE_ROW_OUT)
+ .setFailureTag(DEADLETTER_OUT)
+ .build());
+
+ Map> storeTags =
+ stores.stream()
+ .map(s -> Pair.of(s, new TupleTag()))
+ .collect(Collectors.toMap(Pair::getLeft, Pair::getRight));
+
+ // Step 4. Allocate validated rows to stores by store subscription
+ PCollectionTuple storeAllocatedRows =
+ validatedRows
+ .get(FEATURE_ROW_OUT)
+ .apply(
+ FeatureRowToStoreAllocator.newBuilder()
+ .setStores(stores)
+ .setStoreTags(storeTags)
+ .build());
for (Store store : stores) {
- List subscribedFeatureSets =
- SpecUtil.getSubscribedFeatureSets(store.getSubscriptionsList(), featureSets);
-
- // Generate tags by key
- Map featureSetSpecsByKey = new HashMap<>();
- subscribedFeatureSets.stream()
- .forEach(
- fs -> {
- String ref = getFeatureSetReference(fs.getSpec());
- featureSetSpecsByKey.put(ref, fs.getSpec());
- });
-
- PCollection> staticSpecs =
- pipeline
- .apply(Create.of(""))
- .apply(
- "TemporarySpecSource",
- ParDo.of(
- new DoFn>() {
- @ProcessElement
- public void process(ProcessContext c) {
- featureSetSpecsByKey.forEach((key, value) -> c.output(KV.of(key, value)));
- }
- }));
-
- FeatureSink featureSink = getFeatureSink(store, staticSpecs);
-
- // TODO: make the source part of the job initialisation options
- Source source = subscribedFeatureSets.get(0).getSpec().getSource();
-
- for (FeatureSet featureSet : subscribedFeatureSets) {
- // Ensure Store has valid configuration and Feast can access it.
- featureSink.prepareWrite(featureSet);
- }
+ FeatureSink featureSink = getFeatureSink(store, featureSetSpecs);
- // Step 1. Read messages from Feast Source as FeatureRow.
- PCollectionTuple convertedFeatureRows =
- pipeline.apply(
- "ReadFeatureRowFromSource",
- ReadFromSource.newBuilder()
- .setSource(source)
- .setSuccessTag(FEATURE_ROW_OUT)
- .setFailureTag(DEADLETTER_OUT)
- .build());
-
- // Step 2. Process and validate incoming FeatureRows
- PCollectionTuple validatedRows =
- convertedFeatureRows
- .get(FEATURE_ROW_OUT)
- .apply(
- ProcessAndValidateFeatureRows.newBuilder()
- .setDefaultProject(options.getDefaultFeastProject())
- .setFeatureSetSpecs(featureSetSpecsByKey)
- .setSuccessTag(FEATURE_ROW_OUT)
- .setFailureTag(DEADLETTER_OUT)
- .build());
-
- // Step 3. Write FeatureRow to the corresponding Store.
+ // Step 5. Write FeatureRow to the corresponding Store.
WriteResult writeFeatureRows =
- validatedRows.get(FEATURE_ROW_OUT).apply("WriteFeatureRowToStore", featureSink.writer());
+ storeAllocatedRows
+ .get(storeTags.get(store))
+ .apply("WriteFeatureRowToStore", featureSink.writer());
- // Step 4. Write FailedElements to a dead letter table in BigQuery.
+ // Step 6. Write FailedElements to a dead letter table in BigQuery.
if (options.getDeadLetterTableSpec() != null) {
// TODO: make deadletter destination type configurable
DeadletterSink deadletterSink =
@@ -172,7 +172,7 @@ public void process(ProcessContext c) {
.apply("WriteFailedElements_WriteFeatureRowToStore", deadletterSink.write());
}
- // Step 5. Write metrics to a metrics sink.
+ // Step 7. Write metrics to a metrics sink.
writeFeatureRows
.getSuccessfulInserts()
.apply("WriteSuccessMetrics", WriteSuccessMetricsTransform.create(store.getName()));
@@ -182,6 +182,13 @@ public void process(ProcessContext c) {
.apply("WriteFailureMetrics", WriteFailureMetricsTransform.create(store.getName()));
}
+ // Step 8. Send ack that FeatureSetSpec state is updated
+ featureSetSpecs.apply(
+ "WriteAck",
+ WriteFeatureSetSpecAck.newBuilder()
+ .setSpecsStreamingUpdateConfig(specsStreamingUpdateConfig)
+ .build());
+
return pipeline.run();
}
}
diff --git a/ingestion/src/main/java/feast/ingestion/options/ImportOptions.java b/ingestion/src/main/java/feast/ingestion/options/ImportOptions.java
index e3a1b841c4a..9ee1ff891af 100644
--- a/ingestion/src/main/java/feast/ingestion/options/ImportOptions.java
+++ b/ingestion/src/main/java/feast/ingestion/options/ImportOptions.java
@@ -36,16 +36,29 @@ public interface ImportOptions extends PipelineOptions, DataflowPipelineOptions,
@Required
@Description(
- "JSON string representation of the FeatureSet that the import job will process, in BZip2 binary format."
- + "FeatureSet follows the format in feast.core.FeatureSet proto."
- + "Mutliple FeatureSetSpec can be passed by specifying '--featureSet={...}' multiple times"
+ "JSON string representation of the SpecsPipe configuration."
+ + "Job will use this to know where read new FeatureSetSpec from (kafka broker & topic)"
+ + "and where send acknowledgment on successful update of job's state to."
+ + "SpecsStreamingUpdateConfig follows the format in feast.core.IngestionJob.SpecsStreamingUpdateConfig proto."
+ "The conversion of Proto message to JSON should follow this mapping:"
+ "https://developers.google.com/protocol-buffers/docs/proto3#json"
+ "Please minify and remove all insignificant whitespace such as newline in the JSON string"
+ "to prevent error when parsing the options")
- byte[] getFeatureSetJson();
+ String getSpecsStreamingUpdateConfigJson();
- void setFeatureSetJson(byte[] featureSetJson);
+ void setSpecsStreamingUpdateConfigJson(String json);
+
+ @Required
+ @Description(
+ "JSON string representation of the Source that will be used to read FeatureRows from."
+ + "Source follows the format in featst.core.Source proto. Currently only kafka source is supported"
+ + "The conversion of Proto message to JSON should follow this mapping:"
+ + "https://developers.google.com/protocol-buffers/docs/proto3#json"
+ + "Please minify and remove all insignificant whitespace such as newline in the JSON string"
+ + "to prevent error when parsing the options")
+ String getSourceJson();
+
+ void setSourceJson(String json);
@Required
@Description(
diff --git a/ingestion/src/main/java/feast/ingestion/transform/FeatureRowToStoreAllocator.java b/ingestion/src/main/java/feast/ingestion/transform/FeatureRowToStoreAllocator.java
new file mode 100644
index 00000000000..48889773c49
--- /dev/null
+++ b/ingestion/src/main/java/feast/ingestion/transform/FeatureRowToStoreAllocator.java
@@ -0,0 +1,93 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 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 static feast.ingestion.utils.SpecUtil.parseFeatureSetReference;
+
+import com.google.auto.value.AutoValue;
+import feast.ingestion.utils.SpecUtil;
+import feast.proto.core.StoreProto;
+import feast.proto.types.FeatureRowProto;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+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;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+import org.apache.commons.lang3.tuple.Pair;
+
+/**
+ * For each incoming {@link FeatureRowProto.FeatureRow} allocator choose only stores that
+ * subscripted to its project and featureSet names.
+ *
+ * Return PCollectionTuple with one {@link TupleTag} per {@link StoreProto.Store}. Tags must be
+ * generated in advance.
+ */
+@AutoValue
+public abstract class FeatureRowToStoreAllocator
+ extends PTransform, PCollectionTuple> {
+ public abstract List getStores();
+
+ public abstract Map> getStoreTags();
+
+ public static Builder newBuilder() {
+ return new AutoValue_FeatureRowToStoreAllocator.Builder();
+ }
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setStores(List stores);
+
+ public abstract Builder setStoreTags(
+ Map> tags);
+
+ public abstract FeatureRowToStoreAllocator build();
+ }
+
+ @Override
+ public PCollectionTuple expand(PCollection input) {
+ return input.apply(
+ "AssignRowToStore",
+ ParDo.of(
+ new DoFn() {
+ @ProcessElement
+ public void process(ProcessContext c, @Element FeatureRowProto.FeatureRow row) {
+ Pair projectAndSetNames =
+ parseFeatureSetReference(row.getFeatureSet());
+ getStores().stream()
+ .filter(
+ s ->
+ SpecUtil.isSubscribedToFeatureSet(
+ s.getSubscriptionsList(),
+ projectAndSetNames.getLeft(),
+ projectAndSetNames.getRight()))
+ .forEach(s -> c.output(getStoreTags().get(s), row));
+ }
+ })
+ .withOutputTags(
+ getStoreTags().get(getStores().get(0)),
+ TupleTagList.of(
+ getStores().stream()
+ .skip(1)
+ .map(getStoreTags()::get)
+ .collect(Collectors.toList()))));
+ }
+}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/ProcessAndValidateFeatureRows.java b/ingestion/src/main/java/feast/ingestion/transform/ProcessAndValidateFeatureRows.java
index 2ce8918ccf9..729bff69a7f 100644
--- a/ingestion/src/main/java/feast/ingestion/transform/ProcessAndValidateFeatureRows.java
+++ b/ingestion/src/main/java/feast/ingestion/transform/ProcessAndValidateFeatureRows.java
@@ -19,25 +19,20 @@
import com.google.auto.value.AutoValue;
import feast.ingestion.transform.fn.ProcessFeatureRowDoFn;
import feast.ingestion.transform.fn.ValidateFeatureRowDoFn;
-import feast.ingestion.values.FeatureSet;
import feast.proto.core.FeatureSetProto;
import feast.proto.types.FeatureRowProto.FeatureRow;
import feast.storage.api.writer.FailedElement;
import java.util.Map;
-import java.util.stream.Collectors;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PCollectionTuple;
-import org.apache.beam.sdk.values.TupleTag;
-import org.apache.beam.sdk.values.TupleTagList;
-import org.apache.commons.lang3.tuple.Pair;
+import org.apache.beam.sdk.values.*;
@AutoValue
public abstract class ProcessAndValidateFeatureRows
extends PTransform, PCollectionTuple> {
- public abstract Map getFeatureSetSpecs();
+ public abstract PCollectionView>>
+ getFeatureSetSpecs();
public abstract String getDefaultProject();
@@ -53,7 +48,7 @@ public static Builder newBuilder() {
public abstract static class Builder {
public abstract Builder setFeatureSetSpecs(
- Map featureSets);
+ PCollectionView>> featureSets);
public abstract Builder setDefaultProject(String defaultProject);
@@ -66,22 +61,17 @@ public abstract Builder setFeatureSetSpecs(
@Override
public PCollectionTuple expand(PCollection input) {
-
- Map featureSets =
- getFeatureSetSpecs().entrySet().stream()
- .map(e -> Pair.of(e.getKey(), new FeatureSet(e.getValue())))
- .collect(Collectors.toMap(Pair::getLeft, Pair::getRight));
-
return input
.apply("ProcessFeatureRows", ParDo.of(new ProcessFeatureRowDoFn(getDefaultProject())))
.apply(
"ValidateFeatureRows",
ParDo.of(
ValidateFeatureRowDoFn.newBuilder()
- .setFeatureSets(featureSets)
+ .setFeatureSets(getFeatureSetSpecs())
.setSuccessTag(getSuccessTag())
.setFailureTag(getFailureTag())
.build())
+ .withSideInputs(getFeatureSetSpecs())
.withOutputTags(getSuccessTag(), TupleTagList.of(getFailureTag())));
}
}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/fn/ValidateFeatureRowDoFn.java b/ingestion/src/main/java/feast/ingestion/transform/fn/ValidateFeatureRowDoFn.java
index 856d94828ad..b89af891407 100644
--- a/ingestion/src/main/java/feast/ingestion/transform/fn/ValidateFeatureRowDoFn.java
+++ b/ingestion/src/main/java/feast/ingestion/transform/fn/ValidateFeatureRowDoFn.java
@@ -17,8 +17,10 @@
package feast.ingestion.transform.fn;
import com.google.auto.value.AutoValue;
+import com.google.common.collect.Iterators;
import feast.ingestion.values.FeatureSet;
import feast.ingestion.values.Field;
+import feast.proto.core.FeatureSetProto;
import feast.proto.types.FeatureRowProto.FeatureRow;
import feast.proto.types.FieldProto;
import feast.proto.types.ValueProto.Value.ValCase;
@@ -27,12 +29,14 @@
import java.util.List;
import java.util.Map;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
@AutoValue
public abstract class ValidateFeatureRowDoFn extends DoFn {
- public abstract Map getFeatureSets();
+ public abstract PCollectionView>>
+ getFeatureSets();
public abstract TupleTag getSuccessTag();
@@ -45,7 +49,8 @@ public static Builder newBuilder() {
@AutoValue.Builder
public abstract static class Builder {
- public abstract Builder setFeatureSets(Map featureSets);
+ public abstract Builder setFeatureSets(
+ PCollectionView>> featureSets);
public abstract Builder setSuccessTag(TupleTag successTag);
@@ -58,9 +63,14 @@ public abstract static class Builder {
public void processElement(ProcessContext context) {
String error = null;
FeatureRow featureRow = context.element();
- FeatureSet featureSet = getFeatureSets().get(featureRow.getFeatureSet());
+ Iterable featureSetSpecs =
+ context.sideInput(getFeatureSets()).get(featureRow.getFeatureSet());
+
List fields = new ArrayList<>();
- if (featureSet != null) {
+ if (featureSetSpecs != null) {
+ FeatureSetProto.FeatureSetSpec latestSpec = Iterators.getLast(featureSetSpecs.iterator());
+ FeatureSet featureSet = new FeatureSet(latestSpec);
+
for (FieldProto.Field field : featureRow.getFieldsList()) {
Field fieldSpec = featureSet.getField(field.getName());
if (fieldSpec == null) {
@@ -98,9 +108,10 @@ public void processElement(ProcessContext context) {
.setJobName(context.getPipelineOptions().getJobName())
.setPayload(featureRow.toString())
.setErrorMessage(error);
- if (featureSet != null) {
- String[] split = featureSet.getReference().split("/");
- failedElement = failedElement.setProjectName(split[0]).setFeatureSetName(split[1]);
+ if (featureSetSpecs != null) {
+ FeatureSetProto.FeatureSetSpec spec = Iterators.getLast(featureSetSpecs.iterator());
+ failedElement =
+ failedElement.setProjectName(spec.getProject()).setFeatureSetName(spec.getName());
}
context.output(getFailureTag(), failedElement.build());
} else {
diff --git a/ingestion/src/main/java/feast/ingestion/transform/specs/FilterRelevantFunction.java b/ingestion/src/main/java/feast/ingestion/transform/specs/FilterRelevantFunction.java
new file mode 100644
index 00000000000..cb2c84e2cd0
--- /dev/null
+++ b/ingestion/src/main/java/feast/ingestion/transform/specs/FilterRelevantFunction.java
@@ -0,0 +1,52 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 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.specs;
+
+import feast.ingestion.utils.SpecUtil;
+import feast.proto.core.FeatureSetProto;
+import feast.proto.core.SourceProto;
+import feast.proto.core.StoreProto;
+import java.util.List;
+import org.apache.beam.sdk.transforms.ProcessFunction;
+import org.apache.beam.sdk.values.KV;
+
+/**
+ * Selects only those FeatureSetSpecs that have the same source as current Job and their featureSet
+ * reference matches to current Job Subscriptions (extracted from Stores)
+ */
+public class FilterRelevantFunction
+ implements ProcessFunction, Boolean> {
+ private final List stores;
+ private final SourceProto.Source source;
+
+ public FilterRelevantFunction(SourceProto.Source source, List stores) {
+ this.source = source;
+ this.stores = stores;
+ }
+
+ @Override
+ public Boolean apply(KV input) throws Exception {
+ return stores.stream()
+ .anyMatch(
+ s ->
+ SpecUtil.isSubscribedToFeatureSet(
+ s.getSubscriptionsList(),
+ input.getValue().getProject(),
+ input.getValue().getName()))
+ && input.getValue().getSource().equals(source);
+ }
+}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/specs/KafkaRecordToFeatureSetSpec.java b/ingestion/src/main/java/feast/ingestion/transform/specs/KafkaRecordToFeatureSetSpec.java
new file mode 100644
index 00000000000..47aefd0c0bb
--- /dev/null
+++ b/ingestion/src/main/java/feast/ingestion/transform/specs/KafkaRecordToFeatureSetSpec.java
@@ -0,0 +1,53 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 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.specs;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import feast.proto.core.FeatureSetProto;
+import org.apache.beam.sdk.io.kafka.KafkaRecord;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.values.KV;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Parse FeatureSetSpec from KafkaRecord. We expect message with:
+ *
+ *
+ *
+ *
+ * - key: String - FeatureSet reference
+ *
- value: FeatureSetSpec serialized in proto
+ *
+ */
+public class KafkaRecordToFeatureSetSpec
+ extends DoFn, KV> {
+ private static final Logger log = LoggerFactory.getLogger(KafkaRecordToFeatureSetSpec.class);
+
+ @ProcessElement
+ public void process(ProcessContext c) {
+ try {
+ FeatureSetProto.FeatureSetSpec featureSetSpec =
+ FeatureSetProto.FeatureSetSpec.parseFrom(c.element().getKV().getValue());
+ c.output(KV.of(new String(c.element().getKV().getKey()), featureSetSpec));
+ } catch (InvalidProtocolBufferException e) {
+ log.error(
+ String.format(
+ "Unable to decode FeatureSetSpec with reference %s", c.element().getKV().getKey()));
+ }
+ }
+}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/specs/ReadFeatureSetSpecs.java b/ingestion/src/main/java/feast/ingestion/transform/specs/ReadFeatureSetSpecs.java
new file mode 100644
index 00000000000..c777438be6a
--- /dev/null
+++ b/ingestion/src/main/java/feast/ingestion/transform/specs/ReadFeatureSetSpecs.java
@@ -0,0 +1,107 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 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.specs;
+
+import com.google.auto.value.AutoValue;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import feast.proto.core.FeatureSetProto.FeatureSetSpec;
+import feast.proto.core.IngestionJobProto;
+import feast.proto.core.SourceProto;
+import feast.proto.core.StoreProto;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import org.apache.beam.sdk.io.kafka.KafkaIO;
+import org.apache.beam.sdk.transforms.*;
+import org.apache.beam.sdk.transforms.windowing.AfterPane;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindows;
+import org.apache.beam.sdk.transforms.windowing.Repeatedly;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.joda.time.Duration;
+
+/**
+ * Source of {@link FeatureSetSpec}
+ *
+ * Reads from Kafka topic (from the beginning) {@link FeatureSetSpec}.
+ *
+ *
Filters only relevant Specs by {@link feast.proto.core.StoreProto.Store} subscriptions and
+ * {@link feast.proto.core.SourceProto.Source}.
+ *
+ *
Compacts {@link FeatureSetSpec} by reference (if it was not yet compacted in Kafka).
+ */
+@AutoValue
+public abstract class ReadFeatureSetSpecs
+ extends PTransform>> {
+ public abstract IngestionJobProto.SpecsStreamingUpdateConfig getSpecsStreamingUpdateConfig();
+
+ public abstract SourceProto.Source getSource();
+
+ public abstract List getStores();
+
+ public static Builder newBuilder() {
+ return new AutoValue_ReadFeatureSetSpecs.Builder();
+ }
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setSpecsStreamingUpdateConfig(
+ IngestionJobProto.SpecsStreamingUpdateConfig config);
+
+ public abstract Builder setSource(SourceProto.Source source);
+
+ public abstract Builder setStores(List stores);
+
+ public abstract ReadFeatureSetSpecs build();
+ }
+
+ @Override
+ public PCollection> expand(PBegin input) {
+ return input
+ .apply(
+ KafkaIO.readBytes()
+ .withBootstrapServers(
+ getSpecsStreamingUpdateConfig().getSource().getBootstrapServers())
+ .withTopic(getSpecsStreamingUpdateConfig().getSource().getTopic())
+ .withConsumerConfigUpdates(
+ ImmutableMap.of(
+ ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
+ "earliest",
+ ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
+ false)))
+ .apply("ParseFeatureSetSpec", ParDo.of(new KafkaRecordToFeatureSetSpec()))
+ .apply("OnlyRelevantSpecs", Filter.by(new FilterRelevantFunction(getSource(), getStores())))
+ .apply(
+ Window.>into(new GlobalWindows())
+ .triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(1)))
+ .accumulatingFiredPanes()
+ .withAllowedLateness(Duration.ZERO))
+ .apply(
+ Combine.perKey(
+ (SerializableFunction, FeatureSetSpec>)
+ specs -> {
+ ArrayList featureSetSpecs = Lists.newArrayList(specs);
+ featureSetSpecs.sort(
+ Comparator.comparing(FeatureSetSpec::getVersion).reversed());
+ return featureSetSpecs.get(0);
+ }));
+ }
+}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/specs/WriteFeatureSetSpecAck.java b/ingestion/src/main/java/feast/ingestion/transform/specs/WriteFeatureSetSpecAck.java
new file mode 100644
index 00000000000..b5d49b70766
--- /dev/null
+++ b/ingestion/src/main/java/feast/ingestion/transform/specs/WriteFeatureSetSpecAck.java
@@ -0,0 +1,85 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 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.specs;
+
+import com.google.auto.value.AutoValue;
+import feast.proto.core.FeatureSetProto;
+import feast.proto.core.IngestionJobProto;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.beam.sdk.io.kafka.KafkaIO;
+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.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+/**
+ * Converts input {@link feast.proto.core.FeatureSetProto.FeatureSetSpec} into {@link
+ * feast.proto.core.IngestionJobProto.FeatureSetSpecAck} message and writes it to kafka (ack-topic).
+ */
+@AutoValue
+public abstract class WriteFeatureSetSpecAck
+ extends PTransform>, PDone> {
+ public abstract IngestionJobProto.SpecsStreamingUpdateConfig getSpecsStreamingUpdateConfig();
+
+ public static Builder newBuilder() {
+ return new AutoValue_WriteFeatureSetSpecAck.Builder();
+ }
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setSpecsStreamingUpdateConfig(
+ IngestionJobProto.SpecsStreamingUpdateConfig config);
+
+ public abstract WriteFeatureSetSpecAck build();
+ }
+
+ @Override
+ public PDone expand(PCollection> input) {
+ return input
+ .apply("FeatureSetSpecToAckMessage", ParDo.of(new BuildAckMessage()))
+ .apply(
+ "ToKafka",
+ KafkaIO.write()
+ .withBootstrapServers(
+ getSpecsStreamingUpdateConfig().getAck().getBootstrapServers())
+ .withTopic(getSpecsStreamingUpdateConfig().getAck().getTopic())
+ .withKeySerializer(StringSerializer.class)
+ .withValueSerializer(ByteArraySerializer.class));
+ }
+
+ private static class BuildAckMessage
+ extends DoFn, KV> {
+ @ProcessElement
+ public void process(ProcessContext c) throws IOException {
+ ByteArrayOutputStream encodedAck = new ByteArrayOutputStream();
+
+ IngestionJobProto.FeatureSetSpecAck.newBuilder()
+ .setFeatureSetReference(c.element().getKey())
+ .setJobName(c.getPipelineOptions().getJobName())
+ .setFeatureSetVersion(c.element().getValue().getVersion())
+ .build()
+ .writeTo(encodedAck);
+
+ c.output(KV.of(c.element().getKey(), encodedAck.toByteArray()));
+ }
+ }
+}
diff --git a/ingestion/src/main/java/feast/ingestion/utils/SpecUtil.java b/ingestion/src/main/java/feast/ingestion/utils/SpecUtil.java
index a1356584b3f..79c81c162d1 100644
--- a/ingestion/src/main/java/feast/ingestion/utils/SpecUtil.java
+++ b/ingestion/src/main/java/feast/ingestion/utils/SpecUtil.java
@@ -20,9 +20,11 @@
import com.google.protobuf.util.JsonFormat;
import feast.ingestion.values.Field;
import feast.proto.core.FeatureSetProto.EntitySpec;
-import feast.proto.core.FeatureSetProto.FeatureSet;
import feast.proto.core.FeatureSetProto.FeatureSetSpec;
import feast.proto.core.FeatureSetProto.FeatureSpec;
+import feast.proto.core.IngestionJobProto;
+import feast.proto.core.IngestionJobProto.SpecsStreamingUpdateConfig;
+import feast.proto.core.SourceProto.Source;
import feast.proto.core.StoreProto.Store;
import feast.proto.core.StoreProto.Store.Subscription;
import java.util.ArrayList;
@@ -30,62 +32,60 @@
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;
+import org.apache.commons.lang3.tuple.Pair;
public class SpecUtil {
+ public static String PROJECT_DEFAULT_NAME = "default";
public static String getFeatureSetReference(FeatureSetSpec featureSetSpec) {
return String.format("%s/%s", featureSetSpec.getProject(), featureSetSpec.getName());
}
+ public static Pair parseFeatureSetReference(String reference) {
+ String[] split = reference.split("/", 2);
+ if (split.length == 1) {
+ return Pair.of(PROJECT_DEFAULT_NAME, split[0]);
+ } else {
+ return Pair.of(split[0], split[1]);
+ }
+ }
+
/** Get only feature set specs that matches the subscription */
- public static List getSubscribedFeatureSets(
- List subscriptions, List featureSets) {
- List subscribed = new ArrayList<>();
- for (FeatureSet featureSet : featureSets) {
- for (Subscription sub : subscriptions) {
- // If configuration missing, fail
- if (sub.getProject().isEmpty() || sub.getName().isEmpty()) {
- throw new IllegalArgumentException(
- String.format("Subscription is missing arguments: %s", sub.toString()));
- }
+ public static boolean isSubscribedToFeatureSet(
+ List subscriptions, String projectName, String featureSetName) {
- // If all wildcards, subscribe to everything
- if (sub.getProject().equals("*") || sub.getName().equals("*")) {
- subscribed.add(featureSet);
- break;
- }
+ for (Subscription sub : subscriptions) {
+ // If configuration missing, fail
+ if (sub.getProject().isEmpty() || sub.getName().isEmpty()) {
+ throw new IllegalArgumentException(
+ String.format("Subscription is missing arguments: %s", sub.toString()));
+ }
- // Match project name
- if (!featureSet.getSpec().getProject().equals(sub.getProject())) {
- continue;
- }
+ // If all wildcards, subscribe to everything
+ if (sub.getProject().equals("*") || sub.getName().equals("*")) {
+ return true;
+ }
- // Convert wildcard to regex
- String subName = sub.getName();
- if (!sub.getName().contains(".*")) {
- subName = subName.replace("*", ".*");
- }
+ // Match project name
+ if (!projectName.equals(sub.getProject())) {
+ continue;
+ }
- // Match feature set name to pattern
- Pattern pattern = Pattern.compile(subName);
- if (!pattern.matcher(featureSet.getSpec().getName()).matches()) {
- continue;
- }
- subscribed.add(featureSet);
+ // Convert wildcard to regex
+ String subName = sub.getName();
+ if (!sub.getName().contains(".*")) {
+ subName = subName.replace("*", ".*");
}
- }
- return subscribed;
- }
- public static List parseFeatureSetSpecJsonList(List jsonList)
- throws InvalidProtocolBufferException {
- List featureSets = new ArrayList<>();
- for (String json : jsonList) {
- FeatureSetSpec.Builder builder = FeatureSetSpec.newBuilder();
- JsonFormat.parser().merge(json, builder);
- featureSets.add(FeatureSet.newBuilder().setSpec(builder.build()).build());
+ // Match feature set name to pattern
+ Pattern pattern = Pattern.compile(subName);
+ if (!pattern.matcher(featureSetName).matches()) {
+ continue;
+ }
+ return true;
}
- return featureSets;
+
+ return false;
}
public static List parseStoreJsonList(List jsonList)
@@ -99,6 +99,19 @@ public static List parseStoreJsonList(List jsonList)
return stores;
}
+ public static Source parseSourceJson(String jsonSource) throws InvalidProtocolBufferException {
+ Source.Builder builder = Source.newBuilder();
+ JsonFormat.parser().merge(jsonSource, builder);
+ return builder.build();
+ }
+
+ public static IngestionJobProto.SpecsStreamingUpdateConfig parseSpecsStreamingUpdateConfig(
+ String jsonConfig) throws InvalidProtocolBufferException {
+ SpecsStreamingUpdateConfig.Builder builder = SpecsStreamingUpdateConfig.newBuilder();
+ JsonFormat.parser().merge(jsonConfig, builder);
+ return builder.build();
+ }
+
public static Map getFieldsByName(FeatureSetSpec featureSetSpec) {
Map fieldByName = new HashMap<>();
for (EntitySpec entitySpec : featureSetSpec.getEntitiesList()) {
diff --git a/ingestion/src/test/java/feast/ingestion/ImportJobTest.java b/ingestion/src/test/java/feast/ingestion/ImportJobTest.java
index 39e42963782..47b5aa3f9b8 100644
--- a/ingestion/src/test/java/feast/ingestion/ImportJobTest.java
+++ b/ingestion/src/test/java/feast/ingestion/ImportJobTest.java
@@ -16,15 +16,18 @@
*/
package feast.ingestion;
+import static feast.ingestion.utils.SpecUtil.getFeatureSetReference;
+
+import com.google.common.collect.ImmutableList;
import com.google.common.io.Files;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.util.JsonFormat;
-import feast.ingestion.options.BZip2Compressor;
import feast.ingestion.options.ImportOptions;
import feast.proto.core.FeatureSetProto.EntitySpec;
import feast.proto.core.FeatureSetProto.FeatureSet;
import feast.proto.core.FeatureSetProto.FeatureSetSpec;
import feast.proto.core.FeatureSetProto.FeatureSpec;
+import feast.proto.core.IngestionJobProto;
import feast.proto.core.SourceProto.KafkaSourceConfig;
import feast.proto.core.SourceProto.Source;
import feast.proto.core.SourceProto.SourceType;
@@ -53,6 +56,7 @@
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import org.apache.beam.repackaged.core.org.apache.commons.lang3.tuple.Pair;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.PipelineResult.State;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
@@ -74,6 +78,9 @@ public class ImportJobTest {
private static final String KAFKA_BOOTSTRAP_SERVERS = KAFKA_HOST + ":" + KAFKA_PORT;
private static final short KAFKA_REPLICATION_FACTOR = 1;
private static final String KAFKA_TOPIC = "topic_1";
+ private static final String KAFKA_SPECS_TOPIC = "topic_specs_1";
+ private static final String KAFKA_SPECS_ACK_TOPIC = "topic_specs_ack_1";
+
private static final long KAFKA_PUBLISH_TIMEOUT_SEC = 10;
@SuppressWarnings("UnstableApiUsage")
@@ -117,6 +124,30 @@ public static void tearDown() {
@Test
public void runPipeline_ShouldWriteToRedisCorrectlyGivenValidSpecAndFeatureRow()
throws IOException, InterruptedException {
+ Source featureSource =
+ Source.newBuilder()
+ .setType(SourceType.KAFKA)
+ .setKafkaSourceConfig(
+ KafkaSourceConfig.newBuilder()
+ .setBootstrapServers(KAFKA_HOST + ":" + KAFKA_PORT)
+ .setTopic(KAFKA_TOPIC)
+ .build())
+ .build();
+
+ IngestionJobProto.SpecsStreamingUpdateConfig specsStreamingUpdateConfig =
+ IngestionJobProto.SpecsStreamingUpdateConfig.newBuilder()
+ .setSource(
+ KafkaSourceConfig.newBuilder()
+ .setBootstrapServers(KAFKA_HOST + ":" + KAFKA_PORT)
+ .setTopic(KAFKA_SPECS_TOPIC)
+ .build())
+ .setAck(
+ KafkaSourceConfig.newBuilder()
+ .setBootstrapServers(KAFKA_HOST + ":" + KAFKA_PORT)
+ .setTopic(KAFKA_SPECS_ACK_TOPIC)
+ .build())
+ .build();
+
FeatureSetSpec spec =
FeatureSetSpec.newBuilder()
.setName("feature_set")
@@ -140,15 +171,7 @@ public void runPipeline_ShouldWriteToRedisCorrectlyGivenValidSpecAndFeatureRow()
FeatureSpec.newBuilder().setName("feature_2").setValueType(Enum.STRING).build())
.addFeatures(
FeatureSpec.newBuilder().setName("feature_3").setValueType(Enum.INT64).build())
- .setSource(
- Source.newBuilder()
- .setType(SourceType.KAFKA)
- .setKafkaSourceConfig(
- KafkaSourceConfig.newBuilder()
- .setBootstrapServers(KAFKA_HOST + ":" + KAFKA_PORT)
- .setTopic(KAFKA_TOPIC)
- .build())
- .build())
+ .setSource(featureSource)
.build();
FeatureSet featureSet = FeatureSet.newBuilder().setSpec(spec).build();
@@ -167,20 +190,16 @@ public void runPipeline_ShouldWriteToRedisCorrectlyGivenValidSpecAndFeatureRow()
.build();
ImportOptions options = PipelineOptionsFactory.create().as(ImportOptions.class);
- BZip2Compressor compressor =
- new BZip2Compressor<>(
- option -> {
- JsonFormat.Printer printer =
- JsonFormat.printer().omittingInsignificantWhitespace().printingEnumsAsInts();
- return printer.print(option).getBytes();
- });
- options.setFeatureSetJson(compressor.compress(spec));
+
+ options.setSpecsStreamingUpdateConfigJson(
+ JsonFormat.printer().print(specsStreamingUpdateConfig));
+ options.setSourceJson(JsonFormat.printer().print(featureSource));
options.setStoreJson(Collections.singletonList(JsonFormat.printer().print(redis)));
options.setDefaultFeastProject("myproject");
options.setProject("");
options.setBlockOnRun(false);
- List input = new ArrayList<>();
+ List> input = new ArrayList<>();
Map expected = new HashMap<>();
LOGGER.info("Generating test data ...");
@@ -189,7 +208,7 @@ public void runPipeline_ShouldWriteToRedisCorrectlyGivenValidSpecAndFeatureRow()
i -> {
FeatureRow randomRow = TestUtil.createRandomFeatureRow(featureSet.getSpec());
RedisKey redisKey = TestUtil.createRedisKey(featureSet.getSpec(), randomRow);
- input.add(randomRow);
+ input.add(Pair.of("", randomRow));
List fields =
randomRow.getFieldsList().stream()
.filter(
@@ -216,7 +235,13 @@ public void runPipeline_ShouldWriteToRedisCorrectlyGivenValidSpecAndFeatureRow()
Assert.assertEquals(pipelineResult.getState(), State.RUNNING);
LOGGER.info("Publishing {} Feature Row messages to Kafka ...", input.size());
- TestUtil.publishFeatureRowsToKafka(
+ TestUtil.publishToKafka(
+ KAFKA_BOOTSTRAP_SERVERS,
+ KAFKA_SPECS_TOPIC,
+ ImmutableList.of(Pair.of(getFeatureSetReference(spec), spec)),
+ ByteArraySerializer.class,
+ KAFKA_PUBLISH_TIMEOUT_SEC);
+ TestUtil.publishToKafka(
KAFKA_BOOTSTRAP_SERVERS,
KAFKA_TOPIC,
input,
diff --git a/ingestion/src/test/java/feast/ingestion/transform/FeatureRowToStoreAllocatorTest.java b/ingestion/src/test/java/feast/ingestion/transform/FeatureRowToStoreAllocatorTest.java
new file mode 100644
index 00000000000..0ba5d563690
--- /dev/null
+++ b/ingestion/src/test/java/feast/ingestion/transform/FeatureRowToStoreAllocatorTest.java
@@ -0,0 +1,99 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 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 com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import feast.proto.core.StoreProto;
+import feast.proto.types.FeatureRowProto;
+import java.util.Map;
+import org.apache.beam.sdk.extensions.protobuf.ProtoCoder;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Count;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.TupleTag;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class FeatureRowToStoreAllocatorTest {
+ @Rule public transient TestPipeline p = TestPipeline.create();
+
+ private StoreProto.Store newStore(String s) {
+ return StoreProto.Store.newBuilder()
+ .addSubscriptions(
+ StoreProto.Store.Subscription.newBuilder().setProject("project").setName(s).build())
+ .build();
+ }
+
+ @Test
+ public void featureRowShouldBeAllocatedToStoreTagsAccordingToSubscription() {
+ StoreProto.Store bqOnlyStore = newStore("bq*");
+ StoreProto.Store redisOnlyStore = newStore("redis*");
+ StoreProto.Store anyStore = newStore("*");
+
+ Map> storeTags =
+ ImmutableMap.of(
+ bqOnlyStore, new TupleTag<>(),
+ redisOnlyStore, new TupleTag<>(),
+ anyStore, new TupleTag<>());
+
+ PCollectionTuple allocatedRows =
+ p.apply(
+ Create.of(
+ FeatureRowProto.FeatureRow.newBuilder().setFeatureSet("project/bq_1").build(),
+ FeatureRowProto.FeatureRow.newBuilder().setFeatureSet("project/bq_2").build(),
+ FeatureRowProto.FeatureRow.newBuilder()
+ .setFeatureSet("project/redis_1")
+ .build(),
+ FeatureRowProto.FeatureRow.newBuilder()
+ .setFeatureSet("project/redis_2")
+ .build(),
+ FeatureRowProto.FeatureRow.newBuilder()
+ .setFeatureSet("project/redis_3")
+ .build()))
+ .apply(
+ FeatureRowToStoreAllocator.newBuilder()
+ .setStoreTags(storeTags)
+ .setStores(ImmutableList.of(bqOnlyStore, redisOnlyStore, anyStore))
+ .build());
+
+ PAssert.that(
+ allocatedRows
+ .get(storeTags.get(bqOnlyStore))
+ .setCoder(ProtoCoder.of(FeatureRowProto.FeatureRow.class))
+ .apply("CountBq", Count.globally()))
+ .containsInAnyOrder(2L);
+
+ PAssert.that(
+ allocatedRows
+ .get(storeTags.get(redisOnlyStore))
+ .setCoder(ProtoCoder.of(FeatureRowProto.FeatureRow.class))
+ .apply("CountRedis", Count.globally()))
+ .containsInAnyOrder(3L);
+
+ PAssert.that(
+ allocatedRows
+ .get(storeTags.get(anyStore))
+ .setCoder(ProtoCoder.of(FeatureRowProto.FeatureRow.class))
+ .apply("CountAll", Count.globally()))
+ .containsInAnyOrder(5L);
+
+ p.run();
+ }
+}
diff --git a/ingestion/src/test/java/feast/ingestion/transform/ProcessAndValidateFeatureRowsTest.java b/ingestion/src/test/java/feast/ingestion/transform/ProcessAndValidateFeatureRowsTest.java
index 8c5d7bd8ed6..010a593ab87 100644
--- a/ingestion/src/test/java/feast/ingestion/transform/ProcessAndValidateFeatureRowsTest.java
+++ b/ingestion/src/test/java/feast/ingestion/transform/ProcessAndValidateFeatureRowsTest.java
@@ -34,7 +34,9 @@
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.View;
import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.junit.Rule;
import org.junit.Test;
@@ -104,6 +106,9 @@ public void shouldWriteSuccessAndFailureTagsCorrectly() {
input.add(FeatureRow.newBuilder().setFeatureSet("invalid").build());
+ PCollectionView>> specsView =
+ p.apply("StaticSpecs", Create.of(featureSetSpecs)).apply(View.asMultimap());
+
PCollectionTuple output =
p.apply(Create.of(input))
.setCoder(ProtoCoder.of(FeatureRow.class))
@@ -112,7 +117,7 @@ public void shouldWriteSuccessAndFailureTagsCorrectly() {
.setDefaultProject("myproject")
.setFailureTag(FAILURE_TAG)
.setSuccessTag(SUCCESS_TAG)
- .setFeatureSetSpecs(featureSetSpecs)
+ .setFeatureSetSpecs(specsView)
.build());
PAssert.that(output.get(SUCCESS_TAG)).containsInAnyOrder(expected);
@@ -154,6 +159,9 @@ public void shouldStripVersions() {
randomRow = randomRow.toBuilder().setFeatureSet("myproject/feature_set:1").build();
input.add(randomRow);
+ PCollectionView>> specsView =
+ p.apply("StaticSpecs", Create.of(featureSetSpecs)).apply(View.asMultimap());
+
PCollectionTuple output =
p.apply(Create.of(input))
.setCoder(ProtoCoder.of(FeatureRow.class))
@@ -162,7 +170,7 @@ public void shouldStripVersions() {
.setDefaultProject("myproject")
.setFailureTag(FAILURE_TAG)
.setSuccessTag(SUCCESS_TAG)
- .setFeatureSetSpecs(featureSetSpecs)
+ .setFeatureSetSpecs(specsView)
.build());
PAssert.that(output.get(SUCCESS_TAG)).containsInAnyOrder(expected);
@@ -203,6 +211,9 @@ public void shouldApplyDefaultProject() {
randomRow = randomRow.toBuilder().setFeatureSet("feature_set").build();
input.add(randomRow);
+ PCollectionView>> specsView =
+ p.apply("StaticSpecs", Create.of(featureSetSpecs)).apply(View.asMultimap());
+
PCollectionTuple output =
p.apply(Create.of(input))
.setCoder(ProtoCoder.of(FeatureRow.class))
@@ -211,7 +222,7 @@ public void shouldApplyDefaultProject() {
.setDefaultProject("myproject")
.setFailureTag(FAILURE_TAG)
.setSuccessTag(SUCCESS_TAG)
- .setFeatureSetSpecs(featureSetSpecs)
+ .setFeatureSetSpecs(specsView)
.build());
PAssert.that(output.get(SUCCESS_TAG)).containsInAnyOrder(expected);
@@ -241,8 +252,8 @@ public void shouldExcludeUnregisteredFields() {
FeatureSpec.newBuilder().setName("feature_2").setValueType(Enum.INT64).build())
.build();
- Map featureSets = new HashMap<>();
- featureSets.put("myproject/feature_set", fs1);
+ Map featureSetSpecs = new HashMap<>();
+ featureSetSpecs.put("myproject/feature_set", fs1);
List input = new ArrayList<>();
List expected = new ArrayList<>();
@@ -258,6 +269,9 @@ public void shouldExcludeUnregisteredFields() {
.setValue(Value.newBuilder().setStringVal("hello")))
.build());
+ PCollectionView>> specsView =
+ p.apply("StaticSpecs", Create.of(featureSetSpecs)).apply(View.asMultimap());
+
PCollectionTuple output =
p.apply(Create.of(input))
.setCoder(ProtoCoder.of(FeatureRow.class))
@@ -266,7 +280,7 @@ public void shouldExcludeUnregisteredFields() {
.setDefaultProject("myproject")
.setFailureTag(FAILURE_TAG)
.setSuccessTag(SUCCESS_TAG)
- .setFeatureSetSpecs(featureSets)
+ .setFeatureSetSpecs(specsView)
.build());
PAssert.that(output.get(SUCCESS_TAG)).containsInAnyOrder(expected);
diff --git a/ingestion/src/test/java/feast/ingestion/transform/specs/FeatureSetSpecReadAndWriteTest.java b/ingestion/src/test/java/feast/ingestion/transform/specs/FeatureSetSpecReadAndWriteTest.java
new file mode 100644
index 00000000000..b0cad59b5e0
--- /dev/null
+++ b/ingestion/src/test/java/feast/ingestion/transform/specs/FeatureSetSpecReadAndWriteTest.java
@@ -0,0 +1,244 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 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.specs;
+
+import static feast.ingestion.utils.SpecUtil.getFeatureSetReference;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import com.google.common.io.Files;
+import com.google.protobuf.InvalidProtocolBufferException;
+import feast.proto.core.FeatureSetProto;
+import feast.proto.core.IngestionJobProto;
+import feast.proto.core.SourceProto;
+import feast.proto.core.StoreProto;
+import feast.test.TestUtil;
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.repackaged.core.org.apache.commons.lang3.tuple.Pair;
+import org.apache.beam.runners.direct.DirectOptions;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.joda.time.Duration;
+import org.junit.*;
+
+public class FeatureSetSpecReadAndWriteTest {
+ @Rule public transient TestPipeline p = TestPipeline.fromOptions(makePipelineOptions());
+
+ private static final String KAFKA_HOST = "localhost";
+ private static final int KAFKA_PORT = 29092;
+ private static final String KAFKA_BOOTSTRAP_SERVERS = KAFKA_HOST + ":" + KAFKA_PORT;
+ private static final short KAFKA_REPLICATION_FACTOR = 1;
+ private static final String KAFKA_TOPIC = "topic";
+ private static final String KAFKA_SPECS_TOPIC = "topic_specs";
+ private static final String KAFKA_SPECS_ACK_TOPIC = "topic_specs_ack";
+
+ private static final long KAFKA_PUBLISH_TIMEOUT_SEC = 10;
+ private static final long KAFKA_POLL_TIMEOUT_SEC = 10;
+
+ private KafkaConsumer consumer;
+
+ @SuppressWarnings("UnstableApiUsage")
+ private static final String ZOOKEEPER_DATA_DIR = Files.createTempDir().getAbsolutePath();
+
+ private static final String ZOOKEEPER_HOST = "localhost";
+ private static final int ZOOKEEPER_PORT = 2183;
+
+ @BeforeClass
+ public static void setupClass() throws IOException, InterruptedException {
+ TestUtil.LocalKafka.start(
+ KAFKA_HOST,
+ KAFKA_PORT,
+ KAFKA_REPLICATION_FACTOR,
+ true,
+ ZOOKEEPER_HOST,
+ ZOOKEEPER_PORT,
+ ZOOKEEPER_DATA_DIR);
+ }
+
+ @Before
+ public void setup() {
+ consumer =
+ TestUtil.makeKafkaConsumer(
+ KAFKA_BOOTSTRAP_SERVERS, KAFKA_SPECS_ACK_TOPIC, AckMessageDeserializer.class);
+ }
+
+ @AfterClass
+ public static void tearDown() {
+ TestUtil.LocalKafka.stop();
+ }
+
+ public static PipelineOptions makePipelineOptions() {
+ DirectOptions options = PipelineOptionsFactory.as(DirectOptions.class);
+ options.setJobName("test_job");
+ options.setBlockOnRun(false);
+ return options;
+ }
+
+ @Test
+ public void pipelineShouldReadSpecsAndAcknowledge() {
+ SourceProto.Source source =
+ SourceProto.Source.newBuilder()
+ .setKafkaSourceConfig(
+ SourceProto.KafkaSourceConfig.newBuilder()
+ .setBootstrapServers(KAFKA_BOOTSTRAP_SERVERS)
+ .setTopic(KAFKA_TOPIC)
+ .build())
+ .build();
+
+ StoreProto.Store store =
+ StoreProto.Store.newBuilder()
+ .addSubscriptions(
+ StoreProto.Store.Subscription.newBuilder()
+ .setProject("project")
+ .setName("*")
+ .build())
+ .build();
+
+ IngestionJobProto.SpecsStreamingUpdateConfig specsStreamingUpdateConfig =
+ IngestionJobProto.SpecsStreamingUpdateConfig.newBuilder()
+ .setSource(
+ SourceProto.KafkaSourceConfig.newBuilder()
+ .setBootstrapServers(KAFKA_BOOTSTRAP_SERVERS)
+ .setTopic(KAFKA_SPECS_TOPIC)
+ .build())
+ .setAck(
+ SourceProto.KafkaSourceConfig.newBuilder()
+ .setBootstrapServers(KAFKA_BOOTSTRAP_SERVERS)
+ .setTopic(KAFKA_SPECS_ACK_TOPIC)
+ .build())
+ .build();
+
+ p.apply(
+ ReadFeatureSetSpecs.newBuilder()
+ .setSource(source)
+ .setStores(ImmutableList.of(store))
+ .setSpecsStreamingUpdateConfig(specsStreamingUpdateConfig)
+ .build())
+ .apply(
+ WriteFeatureSetSpecAck.newBuilder()
+ .setSpecsStreamingUpdateConfig(specsStreamingUpdateConfig)
+ .build());
+
+ // specs' history is being compacted on the initial read
+ publishSpecToKafka("project", "fs", 1, source);
+ publishSpecToKafka("project", "fs", 2, source);
+ publishSpecToKafka("project", "fs", 3, source);
+ publishSpecToKafka("project", "fs_2", 2, source);
+
+ PipelineResult run = p.run();
+ run.waitUntilFinish(Duration.standardSeconds(10));
+
+ List acks = getFeatureSetSpecAcks();
+
+ assertThat(
+ acks,
+ hasItem(
+ IngestionJobProto.FeatureSetSpecAck.newBuilder()
+ .setJobName("test_job")
+ .setFeatureSetVersion(3)
+ .setFeatureSetReference("project/fs")
+ .build()));
+ assertThat(
+ acks,
+ hasItem(
+ IngestionJobProto.FeatureSetSpecAck.newBuilder()
+ .setJobName("test_job")
+ .setFeatureSetVersion(2)
+ .setFeatureSetReference("project/fs_2")
+ .build()));
+
+ // in-flight update 1
+ publishSpecToKafka("project", "fs", 4, source);
+
+ run.waitUntilFinish(Duration.standardSeconds(5));
+
+ assertThat(
+ getFeatureSetSpecAcks(),
+ hasItem(
+ IngestionJobProto.FeatureSetSpecAck.newBuilder()
+ .setJobName("test_job")
+ .setFeatureSetVersion(4)
+ .setFeatureSetReference("project/fs")
+ .build()));
+
+ // in-flight update 2
+ publishSpecToKafka("project", "fs_2", 3, source);
+
+ run.waitUntilFinish(Duration.standardSeconds(5));
+
+ assertThat(
+ getFeatureSetSpecAcks(),
+ hasItem(
+ IngestionJobProto.FeatureSetSpecAck.newBuilder()
+ .setJobName("test_job")
+ .setFeatureSetVersion(3)
+ .setFeatureSetReference("project/fs_2")
+ .build()));
+ }
+
+ private List getFeatureSetSpecAcks() {
+ ConsumerRecords consumerRecords =
+ consumer.poll(java.time.Duration.ofSeconds(KAFKA_POLL_TIMEOUT_SEC));
+
+ return Lists.newArrayList(consumerRecords.records(KAFKA_SPECS_ACK_TOPIC)).stream()
+ .map(ConsumerRecord::value)
+ .collect(Collectors.toList());
+ }
+
+ private void publishSpecToKafka(
+ String project, String name, int version, SourceProto.Source source) {
+ FeatureSetProto.FeatureSetSpec spec =
+ FeatureSetProto.FeatureSetSpec.newBuilder()
+ .setProject(project)
+ .setName(name)
+ .setVersion(version)
+ .setSource(source)
+ .build();
+
+ TestUtil.publishToKafka(
+ KAFKA_BOOTSTRAP_SERVERS,
+ KAFKA_SPECS_TOPIC,
+ ImmutableList.of(Pair.of(getFeatureSetReference(spec), spec)),
+ ByteArraySerializer.class,
+ KAFKA_PUBLISH_TIMEOUT_SEC);
+ }
+
+ public static class AckMessageDeserializer
+ implements Deserializer {
+
+ @Override
+ public IngestionJobProto.FeatureSetSpecAck deserialize(String topic, byte[] data) {
+ try {
+ return IngestionJobProto.FeatureSetSpecAck.parseFrom(data);
+ } catch (InvalidProtocolBufferException e) {
+ e.printStackTrace();
+ return null;
+ }
+ }
+ }
+}
diff --git a/ingestion/src/test/java/feast/ingestion/transform/specs/FilterRelevantTest.java b/ingestion/src/test/java/feast/ingestion/transform/specs/FilterRelevantTest.java
new file mode 100644
index 00000000000..33736dad41d
--- /dev/null
+++ b/ingestion/src/test/java/feast/ingestion/transform/specs/FilterRelevantTest.java
@@ -0,0 +1,109 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 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.specs;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import feast.proto.core.FeatureSetProto;
+import feast.proto.core.SourceProto;
+import feast.proto.core.StoreProto;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Count;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.Filter;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class FilterRelevantTest {
+ @Rule public transient TestPipeline p = TestPipeline.create();
+
+ @Test
+ public void onlySpecsThatMatchStoreSubscriptionShouldPass() {
+ SourceProto.Source source =
+ SourceProto.Source.newBuilder()
+ .setKafkaSourceConfig(
+ SourceProto.KafkaSourceConfig.newBuilder()
+ .setBootstrapServers("localhost")
+ .setTopic("topic")
+ .build())
+ .build();
+
+ StoreProto.Store store1 =
+ StoreProto.Store.newBuilder()
+ .addSubscriptions(
+ StoreProto.Store.Subscription.newBuilder()
+ .setProject("project")
+ .setName("fs*")
+ .build())
+ .build();
+
+ StoreProto.Store store2 =
+ StoreProto.Store.newBuilder()
+ .addSubscriptions(
+ StoreProto.Store.Subscription.newBuilder()
+ .setProject("project_2")
+ .setName("fs_1")
+ .build())
+ .build();
+
+ SourceProto.Source irrelevantSource =
+ SourceProto.Source.newBuilder()
+ .setKafkaSourceConfig(
+ SourceProto.KafkaSourceConfig.newBuilder()
+ .setBootstrapServers("localhost")
+ .setTopic("other_topic")
+ .build())
+ .build();
+
+ PCollection> filtered =
+ p.apply(
+ Create.of(
+ ImmutableMap.of(
+ "project/fs_1", makeSpecBuilder(source).build(), // pass
+ "project/fs_2",
+ makeSpecBuilder(irrelevantSource).build(), // different source
+ "invalid/fs_3",
+ makeSpecBuilder(source)
+ .setProject("invalid")
+ .build(), // invalid project
+ "project/invalid",
+ makeSpecBuilder(source).setName("invalid").build(), // invalid name
+ "project_2/fs_1",
+ makeSpecBuilder(source).setProject("project_2").build() // pass
+ )))
+ .apply(Filter.by(new FilterRelevantFunction(source, ImmutableList.of(store1, store2))));
+
+ PAssert.that(filtered)
+ .containsInAnyOrder(
+ KV.of("project/fs_1", makeSpecBuilder(source).build()),
+ KV.of("project_2/fs_1", makeSpecBuilder(source).setProject("project_2").build()));
+
+ PAssert.that(filtered.apply(Count.globally())).containsInAnyOrder(2L);
+
+ p.run();
+ }
+
+ private FeatureSetProto.FeatureSetSpec.Builder makeSpecBuilder(SourceProto.Source source) {
+ return FeatureSetProto.FeatureSetSpec.newBuilder()
+ .setProject("project")
+ .setName("fs_1")
+ .setSource(source);
+ }
+}
diff --git a/ingestion/src/test/java/feast/test/TestUtil.java b/ingestion/src/test/java/feast/test/TestUtil.java
index 3204d93bcda..91a8b9ba7ea 100644
--- a/ingestion/src/test/java/feast/test/TestUtil.java
+++ b/ingestion/src/test/java/feast/test/TestUtil.java
@@ -18,7 +18,10 @@
import static feast.ingestion.utils.SpecUtil.getFeatureSetReference;
+import com.google.common.collect.ImmutableList;
+import com.google.common.io.Files;
import com.google.protobuf.ByteString;
+import com.google.protobuf.Message;
import com.google.protobuf.util.Timestamps;
import feast.ingestion.transform.metrics.WriteSuccessMetricsTransform;
import feast.proto.core.FeatureSetProto.FeatureSet;
@@ -43,15 +46,20 @@
import java.util.concurrent.TimeoutException;
import kafka.server.KafkaConfig;
import kafka.server.KafkaServerStartable;
+import org.apache.beam.repackaged.core.org.apache.commons.lang3.tuple.Pair;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.PipelineResult.State;
import org.apache.beam.sdk.metrics.MetricResult;
import org.apache.beam.sdk.metrics.MetricResults;
import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.serialization.LongSerializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.zookeeper.server.ServerConfig;
import org.apache.zookeeper.server.ZooKeeperServerMain;
import org.joda.time.Duration;
@@ -113,16 +121,19 @@ public static void start(
kafkaProp.put("zookeeper.connect", zookeeperHost + ":" + zookeeperPort);
kafkaProp.put("host.name", kafkaHost);
kafkaProp.put("port", kafkaPort);
+ kafkaProp.put("log.dir", Files.createTempDir().getAbsolutePath());
kafkaProp.put("offsets.topic.replication.factor", kafkaReplicationFactor);
KafkaConfig kafkaConfig = new KafkaConfig(kafkaProp);
server = new KafkaServerStartable(kafkaConfig);
new Thread(server::startup).start();
+ Thread.sleep(5000);
}
public static void stop() {
if (server != null) {
try {
server.shutdown();
+ LocalZookeeper.stop();
} catch (Exception e) {
e.printStackTrace();
}
@@ -139,23 +150,24 @@ public static void stop() {
* @param valueSerializer in Feast this valueSerializer should be "ByteArraySerializer.class"
* @param publishTimeoutSec duration to wait for publish operation (of each message) to succeed
*/
- public static void publishFeatureRowsToKafka(
+ public static void publishToKafka(
String bootstrapServers,
String topic,
- List messages,
+ List> messages,
Class valueSerializer,
long publishTimeoutSec) {
- Long defaultKey = 1L;
+
Properties prop = new Properties();
- prop.put("bootstrap.servers", bootstrapServers);
- prop.put("key.serializer", LongSerializer.class);
- prop.put("value.serializer", valueSerializer);
- Producer producer = new KafkaProducer<>(prop);
+ prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
+ Producer producer = new KafkaProducer<>(prop);
messages.forEach(
featureRow -> {
- ProducerRecord record =
- new ProducerRecord<>(topic, defaultKey, featureRow.toByteArray());
+ ProducerRecord record =
+ new ProducerRecord<>(
+ topic, featureRow.getLeft(), featureRow.getRight().toByteArray());
try {
producer.send(record).get(publishTimeoutSec, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
@@ -164,6 +176,22 @@ public static void publishFeatureRowsToKafka(
});
}
+ public static KafkaConsumer makeKafkaConsumer(
+ String bootstrapServers, String topic, Class valueDeserializer) {
+ Properties prop = new Properties();
+ prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
+ prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+ prop.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
+
+ KafkaConsumer consumer = new KafkaConsumer<>(prop);
+
+ consumer.subscribe(ImmutableList.of(topic));
+ return consumer;
+ }
+
/**
* Create a Feature Row with random value according to the FeatureSetSpec
*
@@ -315,20 +343,26 @@ public static RedisKey createRedisKey(FeatureSetSpec featureSetSpec, FeatureRow
}
private static class LocalZookeeper {
+ public static Thread thread;
static void start(int zookeeperPort, String zookeeperDataDir) {
- final ZooKeeperServerMain zookeeper = new ZooKeeperServerMain();
+ ZooKeeperServerMain zookeeper = new ZooKeeperServerMain();
final ServerConfig serverConfig = new ServerConfig();
serverConfig.parse(new String[] {String.valueOf(zookeeperPort), zookeeperDataDir});
- new Thread(
+ thread =
+ new Thread(
() -> {
try {
zookeeper.runFromConfig(serverConfig);
- } catch (IOException e) {
+ } catch (Exception e) {
e.printStackTrace();
}
- })
- .start();
+ });
+ thread.start();
+ }
+
+ static void stop() {
+ thread.interrupt();
}
}
diff --git a/protos/feast/core/FeatureSet.proto b/protos/feast/core/FeatureSet.proto
index b0b15276bcf..de22388ce76 100644
--- a/protos/feast/core/FeatureSet.proto
+++ b/protos/feast/core/FeatureSet.proto
@@ -63,6 +63,10 @@ message FeatureSetSpec {
// User defined metadata
map labels = 8;
+
+ // Read-only self-incrementing version that increases monotonically
+ // when changes are made to a feature set
+ int32 version = 9;
}
message EntitySpec {
@@ -150,3 +154,8 @@ enum FeatureSetStatus {
STATUS_JOB_STARTING = 3;
STATUS_READY = 2;
}
+
+enum FeatureSetJobDeliveryStatus {
+ STATUS_IN_PROGRESS = 0;
+ STATUS_DELIVERED = 1;
+}
\ No newline at end of file
diff --git a/protos/feast/core/IngestionJob.proto b/protos/feast/core/IngestionJob.proto
index c63e573ffea..0ce0f9681c0 100644
--- a/protos/feast/core/IngestionJob.proto
+++ b/protos/feast/core/IngestionJob.proto
@@ -63,3 +63,18 @@ enum IngestionJobStatus {
// job has been suspended
SUSPENDED = 8;
}
+
+// Config for bi-directional communication channel between Core Service and Ingestion Job
+message SpecsStreamingUpdateConfig {
+ // out-channel for publishing new FeatureSetSpecs (by Core).
+ // IngestionJob use it as source of existing FeatureSetSpecs and new real-time updates
+ feast.core.KafkaSourceConfig source = 1;
+ // ack-channel for sending acknowledgments when new FeatureSetSpecs is installed in Job
+ feast.core.KafkaSourceConfig ack = 2;
+}
+
+message FeatureSetSpecAck {
+ string feature_set_reference = 1;
+ int32 feature_set_version = 2;
+ string job_name = 3;
+}
\ No newline at end of file
diff --git a/storage/connectors/bigquery/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoadsWithResult.java b/storage/connectors/bigquery/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoadsWithResult.java
index 89ce54b8219..e1c50b62cae 100644
--- a/storage/connectors/bigquery/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoadsWithResult.java
+++ b/storage/connectors/bigquery/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoadsWithResult.java
@@ -9,6 +9,7 @@
import java.util.List;
import java.util.Set;
import javax.annotation.Nullable;
+
import org.apache.beam.sdk.coders.*;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.*;