From 2ba36b7e3ca9efaf0885b240b953bff437263a49 Mon Sep 17 00:00:00 2001 From: Willem Pienaar Date: Mon, 13 Jul 2020 14:43:08 +0800 Subject: [PATCH 1/6] Add SNAPSHOT back to version --- infra/charts/feast/Chart.yaml | 2 +- infra/charts/feast/README.md | 10 +++++----- infra/charts/feast/charts/feast-core/Chart.yaml | 2 +- infra/charts/feast/charts/feast-core/README.md | 2 +- infra/charts/feast/charts/feast-jupyter/Chart.yaml | 2 +- infra/charts/feast/charts/feast-jupyter/README.md | 2 +- infra/charts/feast/charts/feast-serving/Chart.yaml | 2 +- infra/charts/feast/charts/feast-serving/README.md | 2 +- infra/charts/feast/requirements.yaml | 8 ++++---- pom.xml | 2 +- 10 files changed, 17 insertions(+), 17 deletions(-) diff --git a/infra/charts/feast/Chart.yaml b/infra/charts/feast/Chart.yaml index f3a1417db28..29365a760f9 100644 --- a/infra/charts/feast/Chart.yaml +++ b/infra/charts/feast/Chart.yaml @@ -1,4 +1,4 @@ apiVersion: v1 description: Feature store for machine learning. name: feast -version: 0.6.0 +version: 0.6.1-SNAPSHOT diff --git a/infra/charts/feast/README.md b/infra/charts/feast/README.md index fe1141811a5..4bae7a43861 100644 --- a/infra/charts/feast/README.md +++ b/infra/charts/feast/README.md @@ -1,7 +1,7 @@ feast ===== -Feature store for machine learning. Current chart version is `0.6.0` +Feature store for machine learning. Current chart version is `0.6.1-SNAPSHOT` ## TL;DR; @@ -32,10 +32,10 @@ This chart install Feast deployment on a Kubernetes cluster using the [Helm](htt | Repository | Name | Version | |------------|------|---------| -| | feast-core | 0.6.0 | -| | feast-jupyter | 0.6.0 | -| | feast-serving | 0.6.0 | -| | feast-serving | 0.6.0 | +| | feast-core | 0.6.1-SNAPSHOT | +| | feast-jupyter | 0.6.1-SNAPSHOT | +| | feast-serving | 0.6.1-SNAPSHOT | +| | feast-serving | 0.6.1-SNAPSHOT | | | prometheus-statsd-exporter | 0.1.2 | | https://kubernetes-charts-incubator.storage.googleapis.com/ | kafka | 0.20.8 | | https://kubernetes-charts.storage.googleapis.com/ | grafana | 5.0.5 | diff --git a/infra/charts/feast/charts/feast-core/Chart.yaml b/infra/charts/feast/charts/feast-core/Chart.yaml index c90d8fbe061..7dedc54465b 100644 --- a/infra/charts/feast/charts/feast-core/Chart.yaml +++ b/infra/charts/feast/charts/feast-core/Chart.yaml @@ -1,4 +1,4 @@ apiVersion: v1 description: Feast Core registers feature specifications and manage ingestion jobs. name: feast-core -version: 0.6.0 +version: 0.6.1-SNAPSHOT diff --git a/infra/charts/feast/charts/feast-core/README.md b/infra/charts/feast/charts/feast-core/README.md index 082a2b68b91..dea2a02b6c2 100644 --- a/infra/charts/feast/charts/feast-core/README.md +++ b/infra/charts/feast/charts/feast-core/README.md @@ -2,7 +2,7 @@ feast-core ========== Feast Core registers feature specifications and manage ingestion jobs. -Current chart version is `0.6.0` +Current chart version is `0.6.1-SNAPSHOT` diff --git a/infra/charts/feast/charts/feast-jupyter/Chart.yaml b/infra/charts/feast/charts/feast-jupyter/Chart.yaml index b86a3a527a3..1a72c244a59 100644 --- a/infra/charts/feast/charts/feast-jupyter/Chart.yaml +++ b/infra/charts/feast/charts/feast-jupyter/Chart.yaml @@ -1,4 +1,4 @@ apiVersion: v1 description: Feast Jupyter provides a Jupyter server with pre-installed Feast SDK name: feast-jupyter -version: 0.6.0 +version: 0.6.1-SNAPSHOT diff --git a/infra/charts/feast/charts/feast-jupyter/README.md b/infra/charts/feast/charts/feast-jupyter/README.md index 113842f2b25..2499200751e 100644 --- a/infra/charts/feast/charts/feast-jupyter/README.md +++ b/infra/charts/feast/charts/feast-jupyter/README.md @@ -2,7 +2,7 @@ feast-jupyter ============= Feast Jupyter provides a Jupyter server with pre-installed Feast SDK -Current chart version is `0.6.0` +Current chart version is `0.6.1-SNAPSHOT` diff --git a/infra/charts/feast/charts/feast-serving/Chart.yaml b/infra/charts/feast/charts/feast-serving/Chart.yaml index 2b47d359685..9387774ba4b 100644 --- a/infra/charts/feast/charts/feast-serving/Chart.yaml +++ b/infra/charts/feast/charts/feast-serving/Chart.yaml @@ -1,4 +1,4 @@ apiVersion: v1 description: Feast Serving serves low-latency latest features and historical batch features. name: feast-serving -version: 0.6.0 +version: 0.6.1-SNAPSHOT diff --git a/infra/charts/feast/charts/feast-serving/README.md b/infra/charts/feast/charts/feast-serving/README.md index 7b2df780f66..5ff562e6e57 100644 --- a/infra/charts/feast/charts/feast-serving/README.md +++ b/infra/charts/feast/charts/feast-serving/README.md @@ -2,7 +2,7 @@ feast-serving ============= Feast Serving serves low-latency latest features and historical batch features. -Current chart version is `0.6.0` +Current chart version is `0.6.1-SNAPSHOT` diff --git a/infra/charts/feast/requirements.yaml b/infra/charts/feast/requirements.yaml index 57a88d9f0d4..b7ea8c905a5 100644 --- a/infra/charts/feast/requirements.yaml +++ b/infra/charts/feast/requirements.yaml @@ -1,17 +1,17 @@ dependencies: - name: feast-core - version: 0.6.0 + version: 0.6.1-SNAPSHOT condition: feast-core.enabled - name: feast-serving alias: feast-online-serving - version: 0.6.0 + version: 0.6.1-SNAPSHOT condition: feast-online-serving.enabled - name: feast-serving alias: feast-batch-serving - version: 0.6.0 + version: 0.6.1-SNAPSHOT condition: feast-batch-serving.enabled - name: feast-jupyter - version: 0.6.0 + version: 0.6.1-SNAPSHOT condition: feast-jupyter.enabled - name: postgresql version: 8.6.1 diff --git a/pom.xml b/pom.xml index 404b4c33a06..e4d601d1f3d 100644 --- a/pom.xml +++ b/pom.xml @@ -41,7 +41,7 @@ - 0.6.0 + 0.6.1-SNAPSHOT https://github.com/feast-dev/feast UTF-8 From daf2041c1b6211b72f4d96b62514247c212b80c2 Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Mon, 13 Jul 2020 13:08:05 +0300 Subject: [PATCH 2/6] Improve parallelization in Redis Sink (#866) * replace group with batch * featureReference as key * configurable flush frequncy in redis sink * pull config to RedisSink --- infra/scripts/setup-common-functions.sh | 2 +- .../scripts/test-end-to-end-redis-cluster.sh | 1 + infra/scripts/test-end-to-end.sh | 16 ++++++- protos/feast/core/Store.proto | 4 ++ .../redis/writer/RedisCustomIO.java | 43 +++++++------------ .../redis/writer/RedisFeatureSink.java | 23 +++++++++- .../writer/RedisClusterFeatureSinkTest.java | 3 -- .../redis/writer/RedisFeatureSinkTest.java | 6 --- 8 files changed, 57 insertions(+), 41 deletions(-) diff --git a/infra/scripts/setup-common-functions.sh b/infra/scripts/setup-common-functions.sh index 44d900b1a9d..01bc21be224 100755 --- a/infra/scripts/setup-common-functions.sh +++ b/infra/scripts/setup-common-functions.sh @@ -98,7 +98,7 @@ start_feast_serving() { if [ -n "$1" ]; then echo "Custom Spring application.yml location provided: $1" - export CONFIG_ARG="--spring.config.location=file://$1" + export CONFIG_ARG="--spring.config.location=classpath:/application.yml,file://$1" fi nohup java -jar serving/target/feast-serving-$FEAST_BUILD_VERSION.jar $CONFIG_ARG &>/var/log/feast-serving-online.log & diff --git a/infra/scripts/test-end-to-end-redis-cluster.sh b/infra/scripts/test-end-to-end-redis-cluster.sh index ed6cd42bc05..9094fc3a2e0 100755 --- a/infra/scripts/test-end-to-end-redis-cluster.sh +++ b/infra/scripts/test-end-to-end-redis-cluster.sh @@ -56,6 +56,7 @@ feast: config: # Connection string specifies the IP and ports of Redis instances in Redis cluster connection_string: "localhost:7000,localhost:7001,localhost:7002,localhost:7003,localhost:7004,localhost:7005" + flush_frequency_seconds: 1 # Subscriptions indicate which feature sets needs to be retrieved and used to populate this store subscriptions: # Wildcards match all options. No filtering is done. diff --git a/infra/scripts/test-end-to-end.sh b/infra/scripts/test-end-to-end.sh index 6e969bc7fc2..75bacd3560a 100755 --- a/infra/scripts/test-end-to-end.sh +++ b/infra/scripts/test-end-to-end.sh @@ -74,7 +74,21 @@ if [[ ${ENABLE_AUTH} = "True" ]]; start_feast_core fi -start_feast_serving +cat < /tmp/serving.warehouse.application.yml +feast: + stores: + - name: online + type: REDIS + config: + host: localhost + port: 6379 + flush_frequency_seconds: 1 + subscriptions: + - name: "*" + project: "*" +EOF + +start_feast_serving /tmp/serving.warehouse.application.yml install_python_with_miniconda_and_feast_sdk print_banner "Running end-to-end tests with pytest at 'tests/e2e'" diff --git a/protos/feast/core/Store.proto b/protos/feast/core/Store.proto index 2ad90c61e7a..3b4394150db 100644 --- a/protos/feast/core/Store.proto +++ b/protos/feast/core/Store.proto @@ -108,6 +108,8 @@ message Store { int32 initial_backoff_ms = 3; // Optional. Maximum total number of retries for connecting to Redis. Default to zero retries. int32 max_retries = 4; + // Optional. how often flush data to redis + int32 flush_frequency_seconds = 5; } message BigQueryConfig { @@ -129,6 +131,8 @@ message Store { string connection_string = 1; int32 initial_backoff_ms = 2; int32 max_retries = 3; + // Optional. how often flush data to redis + int32 flush_frequency_seconds = 4; } message Subscription { diff --git a/storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisCustomIO.java b/storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisCustomIO.java index ae16a464a28..dcd2e5bfda1 100644 --- a/storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisCustomIO.java +++ b/storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisCustomIO.java @@ -40,14 +40,12 @@ import org.apache.beam.sdk.values.*; import org.apache.commons.lang3.exception.ExceptionUtils; import org.apache.commons.lang3.tuple.ImmutablePair; +import org.joda.time.Duration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class RedisCustomIO { - private static final int DEFAULT_BATCH_SIZE = 1000; - private static final int DEFAULT_TIMEOUT = 2000; - private static TupleTag successfulInsertsTag = new TupleTag("successfulInserts") {}; private static TupleTag failedInsertsTupleTag = @@ -69,7 +67,7 @@ public static class Write extends PTransform, WriteResul private PCollectionView>> featureSetSpecs; private RedisIngestionClient redisIngestionClient; private int batchSize; - private int timeout; + private Duration flushFrequency; public Write( RedisIngestionClient redisIngestionClient, @@ -83,8 +81,8 @@ public Write withBatchSize(int batchSize) { return this; } - public Write withTimeout(int timeout) { - this.timeout = timeout; + public Write withFlushFrequency(Duration frequency) { + this.flushFrequency = frequency; return this; } @@ -92,14 +90,19 @@ public Write withTimeout(int timeout) { public WriteResult expand(PCollection input) { PCollectionTuple redisWrite = input + .apply("FixedFlushWindow", Window.into(FixedWindows.of(flushFrequency))) .apply( - "CollectBatchBeforeWrite", - Window.into(new GlobalWindows()) - .triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(batchSize))) - .discardingFiredPanes()) - .apply("AttachSingletonKey", WithKeys.of((Void) null)) - .apply("GroupOntoSingleton", GroupByKey.create()) + "AttachFeatureReferenceKey", + ParDo.of( + new DoFn>() { + @ProcessElement + public void process(ProcessContext c) { + c.output(KV.of(c.element().getFeatureSet(), c.element())); + } + })) + .apply("IntoBatches", GroupIntoBatches.ofSize(batchSize)) .apply("ExtractResultValues", Values.create()) + .apply("GlobalWindow", Window.>into(new GlobalWindows())) .apply( ParDo.of(new WriteDoFn(redisIngestionClient, featureSetSpecs)) .withOutputTags(successfulInsertsTag, TupleTagList.of(failedInsertsTupleTag)) @@ -112,8 +115,6 @@ public WriteResult expand(PCollection input) { public static class WriteDoFn extends DoFn, FeatureRow> { private PCollectionView>> featureSetSpecsView; - private int batchSize = DEFAULT_BATCH_SIZE; - private int timeout = DEFAULT_TIMEOUT; private RedisIngestionClient redisIngestionClient; WriteDoFn( @@ -124,20 +125,6 @@ public static class WriteDoFn extends DoFn, FeatureRow> { this.featureSetSpecsView = featureSetSpecsView; } - public WriteDoFn withBatchSize(int batchSize) { - if (batchSize > 0) { - this.batchSize = batchSize; - } - return this; - } - - public WriteDoFn withTimeout(int timeout) { - if (timeout > 0) { - this.timeout = timeout; - } - return this; - } - @Setup public void setup() { this.redisIngestionClient.setup(); diff --git a/storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisFeatureSink.java b/storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisFeatureSink.java index 4e3964dd901..6997cbcb877 100644 --- a/storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisFeatureSink.java +++ b/storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisFeatureSink.java @@ -35,9 +35,12 @@ import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionView; +import org.joda.time.Duration; @AutoValue public abstract class RedisFeatureSink implements FeatureSink { + private static final int DEFAULT_BATCH_SIZE = 10000; + private static final int DEFAULT_FREQUENCY_SECONDS = 30; /** * Initialize a {@link RedisFeatureSink.Builder} from a {@link StoreProto.Store.RedisConfig}. @@ -112,12 +115,28 @@ public PCollection prepareWrite( @Override public PTransform, WriteResult> writer() { + int flushFrequencySeconds = DEFAULT_FREQUENCY_SECONDS; + if (getRedisClusterConfig() != null) { + + if (getRedisClusterConfig().getFlushFrequencySeconds() > 0) { + flushFrequencySeconds = getRedisClusterConfig().getFlushFrequencySeconds(); + } + return new RedisCustomIO.Write( - new RedisClusterIngestionClient(getRedisClusterConfig()), getSpecsView()); + new RedisClusterIngestionClient(getRedisClusterConfig()), getSpecsView()) + .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds)) + .withBatchSize(DEFAULT_BATCH_SIZE); + } else if (getRedisConfig() != null) { + if (getRedisConfig().getFlushFrequencySeconds() > 0) { + flushFrequencySeconds = getRedisConfig().getFlushFrequencySeconds(); + } + return new RedisCustomIO.Write( - new RedisStandaloneIngestionClient(getRedisConfig()), getSpecsView()); + new RedisStandaloneIngestionClient(getRedisConfig()), getSpecsView()) + .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds)) + .withBatchSize(DEFAULT_BATCH_SIZE); } else { throw new RuntimeException( "At least one RedisConfig or RedisClusterConfig must be provided to Redis Sink"); diff --git a/storage/connectors/redis/src/test/java/feast/storage/connectors/redis/writer/RedisClusterFeatureSinkTest.java b/storage/connectors/redis/src/test/java/feast/storage/connectors/redis/writer/RedisClusterFeatureSinkTest.java index e9130627b73..2adf0cec47f 100644 --- a/storage/connectors/redis/src/test/java/feast/storage/connectors/redis/writer/RedisClusterFeatureSinkTest.java +++ b/storage/connectors/redis/src/test/java/feast/storage/connectors/redis/writer/RedisClusterFeatureSinkTest.java @@ -33,7 +33,6 @@ import feast.proto.types.FieldProto.Field; import feast.proto.types.ValueProto.Value; import feast.proto.types.ValueProto.ValueType.Enum; -import feast.storage.api.writer.FailedElement; import io.lettuce.core.RedisURI; import io.lettuce.core.cluster.RedisClusterClient; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; @@ -221,7 +220,6 @@ public void shouldRetryFailConnection() throws InterruptedException { p.apply(Create.of(featureRows)) .apply(redisClusterFeatureSink.writer()) .getFailedInserts() - .apply(Window.into(new GlobalWindows()).triggering(Never.ever())) .apply(Count.globally()); redisCluster.stop(); @@ -283,7 +281,6 @@ public void shouldProduceFailedElementIfRetryExceeded() { p.apply(Create.of(featureRows)) .apply("modifiedSink", redisClusterFeatureSink.writer()) .getFailedInserts() - .apply(Window.into(new GlobalWindows()).triggering(Never.ever())) .apply(Count.globally()); PAssert.that(failedElementCount).containsInAnyOrder(1L); diff --git a/storage/connectors/redis/src/test/java/feast/storage/connectors/redis/writer/RedisFeatureSinkTest.java b/storage/connectors/redis/src/test/java/feast/storage/connectors/redis/writer/RedisFeatureSinkTest.java index 27c38e7c69a..63ec136c5d0 100644 --- a/storage/connectors/redis/src/test/java/feast/storage/connectors/redis/writer/RedisFeatureSinkTest.java +++ b/storage/connectors/redis/src/test/java/feast/storage/connectors/redis/writer/RedisFeatureSinkTest.java @@ -34,7 +34,6 @@ import feast.proto.types.FieldProto.Field; import feast.proto.types.ValueProto.Value; import feast.proto.types.ValueProto.ValueType.Enum; -import feast.storage.api.writer.FailedElement; import io.lettuce.core.RedisClient; import io.lettuce.core.RedisURI; import io.lettuce.core.api.StatefulRedisConnection; @@ -49,9 +48,6 @@ 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.windowing.GlobalWindows; -import org.apache.beam.sdk.transforms.windowing.Never; -import org.apache.beam.sdk.transforms.windowing.Window; import org.apache.beam.sdk.values.PCollection; import org.junit.After; import org.junit.Before; @@ -212,7 +208,6 @@ public void shouldRetryFailConnection() throws InterruptedException { p.apply(Create.of(featureRows)) .apply(redisFeatureSink.writer()) .getFailedInserts() - .apply(Window.into(new GlobalWindows()).triggering(Never.ever())) .apply(Count.globally()); redis.stop(); @@ -271,7 +266,6 @@ public void shouldProduceFailedElementIfRetryExceeded() { p.apply(Create.of(featureRows)) .apply(redisFeatureSink.writer()) .getFailedInserts() - .apply(Window.into(new GlobalWindows()).triggering(Never.ever())) .apply(Count.globally()); redis.stop(); From 1c056174d7e087dd9c1e52497d37c7f4c4c40547 Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Mon, 13 Jul 2020 13:35:05 +0300 Subject: [PATCH 3/6] Add IngestionId & EventTimestamp to FeatureRowBatch to calculate lag metric correctly (#874) * ingestionId & eventTimestamp in FeatureRowBatch * refactor idx operations using schema * dummy Co-authored-by: Oleksii Moskalenko --- .../compression/FeatureRowsBatch.java | 89 ++++++++++++++----- .../bigquery/writer/BigQuerySinkTest.java | 9 ++ 2 files changed, 78 insertions(+), 20 deletions(-) diff --git a/storage/connectors/bigquery/src/main/java/feast/storage/connectors/bigquery/compression/FeatureRowsBatch.java b/storage/connectors/bigquery/src/main/java/feast/storage/connectors/bigquery/compression/FeatureRowsBatch.java index ee5b1d67bce..fa9cd94c115 100644 --- a/storage/connectors/bigquery/src/main/java/feast/storage/connectors/bigquery/compression/FeatureRowsBatch.java +++ b/storage/connectors/bigquery/src/main/java/feast/storage/connectors/bigquery/compression/FeatureRowsBatch.java @@ -19,6 +19,8 @@ import static feast.proto.types.ValueProto.Value.ValCase.*; import static feast.storage.connectors.bigquery.common.TypeUtil.*; +import com.google.common.collect.ImmutableList; +import com.google.protobuf.Timestamp; import feast.proto.types.FeatureRowProto; import feast.proto.types.FieldProto; import feast.proto.types.ValueProto; @@ -40,6 +42,8 @@ *

getFeatureRows provides reverse transformation */ public class FeatureRowsBatch implements Serializable { + public static final ImmutableList SERVICE_FIELDS = + ImmutableList.of("eventTimestamp", "ingestionId"); private final Schema schema; private String featureSetReference; private List values = new ArrayList<>(); @@ -118,6 +122,12 @@ private Schema inferCommonSchema(Iterable featureRow featureSetReference = row.getFeatureSet(); } })); + + fieldsInOrder.add( + Schema.Field.of("eventTimestamp", Schema.FieldType.array(Schema.FieldType.INT64))); + fieldsInOrder.add( + Schema.Field.of("ingestionId", Schema.FieldType.array(Schema.FieldType.STRING))); + Schema schema = Schema.builder().addFields(fieldsInOrder).build(); schema.setUUID(UUID.randomUUID()); return schema; @@ -132,16 +142,33 @@ private void initValues() { } private void toColumnar(Iterable featureRows) { + int timestampColumnIdx = schema.indexOf("eventTimestamp"); + int ingestionIdColumnIdx = schema.indexOf("ingestionId"); + featureRows.forEach( row -> { - Map rowValues = - row.getFieldsList().stream() - .collect(Collectors.toMap(FieldProto.Field::getName, FieldProto.Field::getValue)); + Map rowValues; + try { + rowValues = + row.getFieldsList().stream() + .collect( + Collectors.toMap(FieldProto.Field::getName, FieldProto.Field::getValue)); + } catch (IllegalStateException e) { + // row contains feature duplicates + // omitting for now + return; + } - IntStream.range(0, schema.getFieldCount()) + schema + .getFieldNames() .forEach( - idx -> { - Schema.Field field = schema.getField(idx); + fieldName -> { + if (SERVICE_FIELDS.contains(fieldName)) { + return; + } + Schema.Field field = schema.getField(fieldName); + int idx = schema.indexOf(fieldName); + if (rowValues.containsKey(field.getName())) { Object o = protoValueToObject(rowValues.get(field.getName())); if (o != null) { @@ -152,6 +179,10 @@ private void toColumnar(Iterable featureRows) { ((List) values.get(idx)).add(defaultValues.get(field.getName())); }); + + // adding service fields + ((List) values.get(timestampColumnIdx)).add(row.getEventTimestamp().getSeconds()); + ((List) values.get(ingestionIdColumnIdx)).add(row.getIngestionId()); }); } @@ -177,27 +208,45 @@ public static FeatureRowsBatch fromRow(Row row) { } public Iterator getFeatureRows() { + int timestampColumnIdx = schema.indexOf("eventTimestamp"); + int ingestionIdColumnIdx = schema.indexOf("ingestionId"); + return IntStream.range(0, ((List) values.get(0)).size()) .parallel() .mapToObj( rowIdx -> FeatureRowProto.FeatureRow.newBuilder() .setFeatureSet(getFeatureSetReference()) + .setEventTimestamp( + Timestamp.newBuilder() + .setSeconds( + (long) + (((List) values.get(timestampColumnIdx)).get(rowIdx))) + .build()) + .setIngestionId( + (String) (((List) values.get(ingestionIdColumnIdx)).get(rowIdx))) .addAllFields( - IntStream.range(0, schema.getFieldCount()) - .mapToObj( - fieldIdx -> - FieldProto.Field.newBuilder() - .setName(schema.getField(fieldIdx).getName()) - .setValue( - objectToProtoValue( - ((List) values.get(fieldIdx)).get(rowIdx), - schemaToProtoTypes.get( - schema - .getField(fieldIdx) - .getType() - .getCollectionElementType()))) - .build()) + schema.getFieldNames().stream() + .map( + fieldName -> { + if (SERVICE_FIELDS.contains(fieldName)) { + return null; + } + int fieldIdx = schema.indexOf(fieldName); + + return FieldProto.Field.newBuilder() + .setName(schema.getField(fieldIdx).getName()) + .setValue( + objectToProtoValue( + ((List) values.get(fieldIdx)).get(rowIdx), + schemaToProtoTypes.get( + schema + .getField(fieldIdx) + .getType() + .getCollectionElementType()))) + .build(); + }) + .filter(Objects::nonNull) .collect(Collectors.toList())) .build()) .iterator(); diff --git a/storage/connectors/bigquery/src/test/java/feast/storage/connectors/bigquery/writer/BigQuerySinkTest.java b/storage/connectors/bigquery/src/test/java/feast/storage/connectors/bigquery/writer/BigQuerySinkTest.java index 3f35c5e4ae9..d58bebf65be 100644 --- a/storage/connectors/bigquery/src/test/java/feast/storage/connectors/bigquery/writer/BigQuerySinkTest.java +++ b/storage/connectors/bigquery/src/test/java/feast/storage/connectors/bigquery/writer/BigQuerySinkTest.java @@ -124,6 +124,11 @@ private FeatureRow generateRow(String featureSet) { FeatureRow.Builder row = FeatureRow.newBuilder() .setFeatureSet(featureSet) + .setEventTimestamp( + com.google.protobuf.Timestamp.newBuilder() + .setSeconds(System.currentTimeMillis() / 1000) + .build()) + .setIngestionId("ingestion-id") .addFields(field("entity", rd.nextInt(), ValueProto.ValueType.Enum.INT64)) .addFields(FieldProto.Field.newBuilder().setName("null_value").build()); @@ -499,6 +504,8 @@ private List dropNullFeature(List input) { r -> FeatureRow.newBuilder() .setFeatureSet(r.getFeatureSet()) + .setIngestionId(r.getIngestionId()) + .setEventTimestamp(r.getEventTimestamp()) .addAllFields(copyFieldsWithout(r, "null_value")) .build()) .collect(Collectors.toList()); @@ -520,6 +527,8 @@ public static List sortFeaturesByName(List rows) { return FeatureRow.newBuilder() .setFeatureSet(row.getFeatureSet()) + .setEventTimestamp(row.getEventTimestamp()) + .setIngestionId(row.getIngestionId()) .addAllFields(fieldsList) .build(); }) From 53f4f3dcf93d474d9d13a71a91d5e9906ac5a8cb Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Mon, 13 Jul 2020 14:28:05 +0300 Subject: [PATCH 4/6] BQ sink produces sample of successful inserts (#875) * feature row batch produces sample * lint Co-authored-by: Willem Pienaar --- .../compression/FeatureRowsBatch.java | 86 +++++++++++-------- .../bigquery/writer/BigQueryWrite.java | 12 ++- .../bigquery/writer/BigQuerySinkTest.java | 38 ++++++++ 3 files changed, 97 insertions(+), 39 deletions(-) diff --git a/storage/connectors/bigquery/src/main/java/feast/storage/connectors/bigquery/compression/FeatureRowsBatch.java b/storage/connectors/bigquery/src/main/java/feast/storage/connectors/bigquery/compression/FeatureRowsBatch.java index fa9cd94c115..1befae221b7 100644 --- a/storage/connectors/bigquery/src/main/java/feast/storage/connectors/bigquery/compression/FeatureRowsBatch.java +++ b/storage/connectors/bigquery/src/main/java/feast/storage/connectors/bigquery/compression/FeatureRowsBatch.java @@ -207,48 +207,58 @@ public static FeatureRowsBatch fromRow(Row row) { return new FeatureRowsBatch(row.getSchema(), row.getValues()); } - public Iterator getFeatureRows() { + private FeatureRowProto.FeatureRow restoreFeatureRow(int rowIdx) { int timestampColumnIdx = schema.indexOf("eventTimestamp"); int ingestionIdColumnIdx = schema.indexOf("ingestionId"); - return IntStream.range(0, ((List) values.get(0)).size()) + return FeatureRowProto.FeatureRow.newBuilder() + .setFeatureSet(getFeatureSetReference()) + .setEventTimestamp( + Timestamp.newBuilder() + .setSeconds((long) (((List) values.get(timestampColumnIdx)).get(rowIdx))) + .build()) + .setIngestionId((String) (((List) values.get(ingestionIdColumnIdx)).get(rowIdx))) + .addAllFields( + schema.getFieldNames().stream() + .map( + fieldName -> { + if (SERVICE_FIELDS.contains(fieldName)) { + return null; + } + int fieldIdx = schema.indexOf(fieldName); + + return FieldProto.Field.newBuilder() + .setName(schema.getField(fieldIdx).getName()) + .setValue( + objectToProtoValue( + ((List) values.get(fieldIdx)).get(rowIdx), + schemaToProtoTypes.get( + schema + .getField(fieldIdx) + .getType() + .getCollectionElementType()))) + .build(); + }) + .filter(Objects::nonNull) + .collect(Collectors.toList())) + .build(); + } + + public Iterator getFeatureRows() { + int featureCount = ((List) values.get(0)).size(); + + return IntStream.range(0, featureCount).parallel().mapToObj(this::restoreFeatureRow).iterator(); + } + + public Iterator getFeatureRowsSample(int maxCount) { + int featureCount = ((List) values.get(0)).size(); + Random rd = new Random(42); + + return IntStream.range(0, featureCount) + .filter(idx -> rd.nextInt(featureCount) < maxCount) .parallel() - .mapToObj( - rowIdx -> - FeatureRowProto.FeatureRow.newBuilder() - .setFeatureSet(getFeatureSetReference()) - .setEventTimestamp( - Timestamp.newBuilder() - .setSeconds( - (long) - (((List) values.get(timestampColumnIdx)).get(rowIdx))) - .build()) - .setIngestionId( - (String) (((List) values.get(ingestionIdColumnIdx)).get(rowIdx))) - .addAllFields( - schema.getFieldNames().stream() - .map( - fieldName -> { - if (SERVICE_FIELDS.contains(fieldName)) { - return null; - } - int fieldIdx = schema.indexOf(fieldName); - - return FieldProto.Field.newBuilder() - .setName(schema.getField(fieldIdx).getName()) - .setValue( - objectToProtoValue( - ((List) values.get(fieldIdx)).get(rowIdx), - schemaToProtoTypes.get( - schema - .getField(fieldIdx) - .getType() - .getCollectionElementType()))) - .build(); - }) - .filter(Objects::nonNull) - .collect(Collectors.toList())) - .build()) + .mapToObj(this::restoreFeatureRow) + .limit(maxCount) .iterator(); } diff --git a/storage/connectors/bigquery/src/main/java/feast/storage/connectors/bigquery/writer/BigQueryWrite.java b/storage/connectors/bigquery/src/main/java/feast/storage/connectors/bigquery/writer/BigQueryWrite.java index a9d5fb6438e..8a9a5995b62 100644 --- a/storage/connectors/bigquery/src/main/java/feast/storage/connectors/bigquery/writer/BigQueryWrite.java +++ b/storage/connectors/bigquery/src/main/java/feast/storage/connectors/bigquery/writer/BigQueryWrite.java @@ -55,6 +55,7 @@ public class BigQueryWrite extends PTransform, WriteResu private static final Duration BIGQUERY_JOB_MAX_EXPECTING_RESULT_TIME = Duration.standardHours(1); private static final int BIGQUERY_MAX_JOB_RETRIES = 20; private static final int DEFAULT_COMPACTION_BATCH_SIZE = 10000; + private static final int MAX_SUCCESSFUL_OUTPUTS_PER_DESTINATION = 1000; private DatasetId destination; private PCollectionView>> schemas; @@ -63,6 +64,7 @@ public class BigQueryWrite extends PTransform, WriteResu private Duration expectingResultTime = BIGQUERY_JOB_MAX_EXPECTING_RESULT_TIME; private BigQueryServices testServices; private int compactionBatchSize = DEFAULT_COMPACTION_BATCH_SIZE; + private int maxSuccessfulOutputs = MAX_SUCCESSFUL_OUTPUTS_PER_DESTINATION; public BigQueryWrite( DatasetId destination, PCollectionView>> schemas) { @@ -90,6 +92,11 @@ public BigQueryWrite withCompactionBatchSize(int batchSize) { return this; } + public BigQueryWrite withMaxSuccessfulOutputs(int maxSuccessfulOutputs) { + this.maxSuccessfulOutputs = maxSuccessfulOutputs; + return this; + } + /** * BigQuery writer 1. choose destination based on featureSetName {@link * FeatureDynamicDestinations} 2. dynamically pull destination's schema from schemas' view 3. @@ -225,7 +232,10 @@ public void process(ProcessContext c) { result .getAll(inputTag) - .forEach(rows -> rows.getFeatureRows().forEachRemaining(c::output)); + .forEach( + rows -> + rows.getFeatureRowsSample(maxSuccessfulOutputs) + .forEachRemaining(c::output)); } })); } diff --git a/storage/connectors/bigquery/src/test/java/feast/storage/connectors/bigquery/writer/BigQuerySinkTest.java b/storage/connectors/bigquery/src/test/java/feast/storage/connectors/bigquery/writer/BigQuerySinkTest.java index d58bebf65be..ba2089414bd 100644 --- a/storage/connectors/bigquery/src/test/java/feast/storage/connectors/bigquery/writer/BigQuerySinkTest.java +++ b/storage/connectors/bigquery/src/test/java/feast/storage/connectors/bigquery/writer/BigQuerySinkTest.java @@ -21,6 +21,7 @@ import static feast.storage.connectors.bigquery.writer.FeatureSetSpecToTableSchema.*; import static org.hamcrest.CoreMatchers.*; import static org.hamcrest.Matchers.containsInAnyOrder; +import static org.hamcrest.Matchers.lessThan; import static org.junit.Assert.*; import static org.mockito.Mockito.*; import static org.mockito.MockitoAnnotations.initMocks; @@ -498,6 +499,30 @@ public void featureRowCompressShouldPackAndUnpackSuccessfully() { p.run(); } + @Test + public void featureRowBatchShouldSampleOnRestore() { + List stream = + IntStream.range(0, 1000) + .mapToObj(i -> generateRow("project/fs")) + .collect(Collectors.toList()); + + PCollection result = + p.apply(Create.of(stream)) + .apply("KV", ParDo.of(new ExtractKV())) + .apply(new CompactFeatureRows(1000)) + .apply(ParDo.of(new FlatMapWithSample(100))) + .apply(Count.globally()); + + PAssert.that(result) + .satisfies( + r -> { + // sample size is within bound of required size + assertThat(Math.abs(r.iterator().next() - 100), lessThan(5L)); + return null; + }); + p.run(); + } + private List dropNullFeature(List input) { return input.stream() .map( @@ -549,4 +574,17 @@ public Table answer(InvocationOnMock invocationOnMock) throws Throwable { return FakeTable.create(mock(BigQuery.class), tableId, tableDefinition); } } + + private static class FlatMapWithSample extends DoFn, FeatureRow> { + private int sampleSize; + + FlatMapWithSample(int sampleSize) { + this.sampleSize = sampleSize; + } + + @ProcessElement + public void process(ProcessContext c) { + c.element().getValue().getFeatureRowsSample(sampleSize).forEachRemaining(c::output); + } + } } From 17473f7b66362e5ff882d9c98ce914aaa9f609b7 Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Thu, 16 Jul 2020 10:39:06 +0300 Subject: [PATCH 5/6] Prevent race condition in BQ sink jobId generation (#877) * fix race condition * [bq] temp file prefix in global window --- .../io/gcp/bigquery/BatchLoadsWithResult.java | 122 ++++++++---------- 1 file changed, 55 insertions(+), 67 deletions(-) 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 764bfc1f54d..f6be75fe13e 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 @@ -5,15 +5,13 @@ import com.google.api.services.bigquery.model.TableRow; import com.google.auto.value.AutoValue; - import java.util.Collections; import java.util.List; import java.util.Set; import javax.annotation.Nullable; - +import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.coders.*; import org.apache.beam.sdk.options.ValueProvider; -import org.apache.beam.sdk.state.*; import org.apache.beam.sdk.transforms.*; import org.apache.beam.sdk.transforms.windowing.*; import org.apache.beam.sdk.values.*; @@ -26,7 +24,7 @@ @AutoValue public abstract class BatchLoadsWithResult extends PTransform< - PCollection>, PCollection>> { + PCollection>, PCollection>> { static final Logger LOG = LoggerFactory.getLogger(BatchLoadsWithResult.class); @VisibleForTesting @@ -134,13 +132,14 @@ public PCollection> expand( input.getWindowingStrategy().getWindowFn() instanceof FixedWindows, "Input to BQ writer must be windowed in advance"); - final PCollection loadJobIdPrefixView = createLoadJobIdPrefixView(input); + final PCollectionView loadJobIdPrefixView = createLoadJobIdPrefixView(input); final PCollectionView tempFilePrefixView = - createTempFilePrefixView(loadJobIdPrefixView); + createTempFilePrefixView(input.getPipeline()); PCollection> results = input - .apply("WindowWithTrigger", + .apply( + "WindowWithTrigger", Window.>configure() .triggering( Repeatedly.forever( @@ -161,8 +160,8 @@ public void process(ProcessContext c) { .apply( "WriteGroupedRecords", ParDo.of( - new WriteGroupedRecordsToFiles<>( - tempFilePrefixView, DEFAULT_MAX_FILE_SIZE, getRowWriterFactory())) + new WriteGroupedRecordsToFiles<>( + tempFilePrefixView, DEFAULT_MAX_FILE_SIZE, getRowWriterFactory())) .withSideInputs(tempFilePrefixView)) .setCoder(WriteBundlesToFiles.ResultCoder.of(getDestinationCoder())); @@ -176,8 +175,7 @@ tempFilePrefixView, DEFAULT_MAX_FILE_SIZE, getRowWriterFactory())) results .apply( Window.>configure() - .triggering(DefaultTrigger.of()) - ) + .triggering(DefaultTrigger.of())) .apply("AttachSingletonKey", WithKeys.of((Void) null)) .setCoder( KvCoder.of( @@ -187,15 +185,15 @@ tempFilePrefixView, DEFAULT_MAX_FILE_SIZE, getRowWriterFactory())) .apply( "WritePartitionTriggered", ParDo.of( - new WritePartition<>( - false, - getDynamicDestinations(), - tempFilePrefixView, - DEFAULT_MAX_FILES_PER_PARTITION, - DEFAULT_MAX_BYTES_PER_PARTITION, - multiPartitionsTag, - singlePartitionTag, - getRowWriterFactory())) + new WritePartition<>( + false, + getDynamicDestinations(), + tempFilePrefixView, + DEFAULT_MAX_FILES_PER_PARTITION, + DEFAULT_MAX_BYTES_PER_PARTITION, + multiPartitionsTag, + singlePartitionTag, + getRowWriterFactory())) .withSideInputs(tempFilePrefixView) .withOutputTags(multiPartitionsTag, TupleTagList.of(singlePartitionTag))); @@ -206,73 +204,60 @@ tempFilePrefixView, DEFAULT_MAX_FILE_SIZE, getRowWriterFactory())) ShardedKeyCoder.of(NullableCoder.of(getDestinationCoder())), ListCoder.of(StringUtf8Coder.of()))); - return writeSinglePartitionWithResult( - partitions.get(singlePartitionTag), loadJobIdPrefixView.apply(View.asSingleton())); + return writeSinglePartitionWithResult(partitions.get(singlePartitionTag), loadJobIdPrefixView); } - private PCollection createLoadJobIdPrefixView( + /** + * Generates one jobId per window only if any feature row was submitted in this window. We need to + * generate exactly one id per window, otherwise SingletonView will fail. + * + * @param input feature Rows + * @return job id generated once per input's window + */ + private PCollectionView createLoadJobIdPrefixView( PCollection> input) { // We generate new JobId per each (input) window // To keep BQ job's name unique // Windowing of this generator is expected to be synchronized with input window // So generated ids can be applied as side input + + String baseName = input.getPipeline().getOptions().getJobName().replaceAll("-", ""); + return input .apply( - "EraseKey", + "EraseKeyAndValue", ParDo.of( - new DoFn, KV>() { + new DoFn, String>() { @ProcessElement public void process(ProcessContext c) { - c.output(KV.of(null, c.element().getValue())); + // we don't need data, only fact of data existing + c.output(""); } })) .apply( - "CreateJobId", - ParDo.of( - new DoFn, String>() { - @StateId("oncePerWindow") - private final StateSpec> oncePerWindow = StateSpecs.set(BooleanCoder.of()); - - @ProcessElement - public void process( - ProcessContext c, - BoundedWindow w, - @StateId("oncePerWindow") SetState oncePerWindow) { - - // if set already contains something - // it means we already generated Id for this window - Boolean empty = oncePerWindow.isEmpty().read(); - if (empty != null && !empty) { - return; - } - - // trying to add to Set and check if it was added - // if true - we won and Id will be generated in current Process - Boolean insertResult = oncePerWindow.addIfAbsent(true).read(); - if (insertResult != null && !insertResult) { - return; - } - - c.output( - String.format( - "beam_load_%s_%s", - c.getPipelineOptions().getJobName().replaceAll("-", ""), - BigQueryHelpers.randomUUIDString())); - - LOG.info("Pane {}, start: {}, last: {}", c.pane().getIndex(), c.pane().isFirst(), c.pane().isLast()); - LOG.info("[BQ] New window {}, {}", c.timestamp(), w.maxTimestamp()); - } - })); + Combine.globally( + (SerializableFunction, String>) + g -> + String.format( + "beam_load_%s_%s", baseName, BigQueryHelpers.randomUUIDString())) + .withoutDefaults()) + .apply("JobIdView", View.asSingleton()); } - private PCollectionView createTempFilePrefixView(final PCollection jobId) { - return jobId + /** + * Generates one global (per all windows) prefix path to store files before load to BQ + * + * @param p Pipeline + * @return view in global window + */ + private PCollectionView createTempFilePrefixView(final Pipeline p) { + return p.apply("CreateGlobalTempPrefix", Create.of("")) .apply( "GetTempFilePrefix", ParDo.of( new DoFn() { @ProcessElement - public void getTempFilePrefix(ProcessContext c, BoundedWindow w) { + public void getTempFilePrefix(ProcessContext c) { String tempLocationRoot; if (getCustomGcsTempLocation() != null) { tempLocationRoot = getCustomGcsTempLocation().get(); @@ -280,8 +265,11 @@ public void getTempFilePrefix(ProcessContext c, BoundedWindow w) { tempLocationRoot = c.getPipelineOptions().getTempLocation(); } String tempLocation = - resolveTempLocation(tempLocationRoot, "BigQueryWriteTemp", c.element()); - LOG.info("[BQ] temp location generated {}, {}", tempLocation, w.maxTimestamp()); + resolveTempLocation( + tempLocationRoot, + "BigQueryWriteTemp", + c.getPipelineOptions().getJobName()); + c.output(tempLocation); } })) From 0dce4f1aaeaa8f150bc4fdd18164872d0f570ea7 Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Fri, 17 Jul 2020 12:34:16 +0300 Subject: [PATCH 6/6] 0.6.1 Release (#888) --- CHANGELOG.md | 14 ++++++++++++++ pom.xml | 2 +- 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 343b3a6fa70..6d86eac1cb8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,19 @@ # Changelog +## [v0.6.1](https://github.com/feast-dev/feast/tree/v0.6.1) (2020-07-17) + +[Full Changelog](https://github.com/feast-dev/feast/compare/v0.6.0...v0.6.1) + +**Implemented enhancements:** + +- Improve parallelization in Redis Sink [\#866](https://github.com/feast-dev/feast/pull/866) ([pyalex](https://github.com/pyalex)) +- BQ sink produces sample of successful inserts [\#875](https://github.com/feast-dev/feast/pull/875) ([pyalex](https://github.com/pyalex)) + +**Fixed bugs:** + +- Add IngestionId & EventTimestamp to FeatureRowBatch to calculate lag metric correctly [\#874](https://github.com/feast-dev/feast/pull/874) ([pyalex](https://github.com/pyalex)) +- Prevent race condition in BQ sink jobId generation [\#877](https://github.com/feast-dev/feast/pull/877) ([pyalex](https://github.com/pyalex)) + ## [v0.6.0](https://github.com/feast-dev/feast/tree/v0.6.0) (2020-07-13) [Full Changelog](https://github.com/feast-dev/feast/compare/v0.5.1...v0.6.0) diff --git a/pom.xml b/pom.xml index e4d601d1f3d..624350d5901 100644 --- a/pom.xml +++ b/pom.xml @@ -41,7 +41,7 @@ - 0.6.1-SNAPSHOT + 0.6.1 https://github.com/feast-dev/feast UTF-8