From 97b0a0bd692b1d84e4827c9f6c38f073aae4ebd6 Mon Sep 17 00:00:00 2001 From: Kevin Zhang Date: Mon, 6 Jun 2022 14:43:08 -0700 Subject: [PATCH 1/7] Add sliding window to aggregations Signed-off-by: Kevin Zhang --- protos/feast/core/Aggregation.proto | 1 + sdk/python/feast/aggregation.py | 22 +++++++++++++++++++++- 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/protos/feast/core/Aggregation.proto b/protos/feast/core/Aggregation.proto index d848ce69721..c2cf4f8f193 100644 --- a/protos/feast/core/Aggregation.proto +++ b/protos/feast/core/Aggregation.proto @@ -11,4 +11,5 @@ message Aggregation { string column = 1; string function = 2; google.protobuf.Duration time_window = 3; + google.protobuf.Duration sliding_window = 4; } \ No newline at end of file diff --git a/sdk/python/feast/aggregation.py b/sdk/python/feast/aggregation.py index 0a5fe845659..d484f876aa3 100644 --- a/sdk/python/feast/aggregation.py +++ b/sdk/python/feast/aggregation.py @@ -14,21 +14,28 @@ class Aggregation: column: str # Column name of the feature we are aggregating. function: str # Provided built in aggregations sum, max, min, count mean time_window: timedelta # The time window for this aggregation. + sliding_window: timedelta # The sliding window for these aggregations """ column: str function: str time_window: Optional[timedelta] + sliding_window: timedelta def __init__( self, column: Optional[str] = "", function: Optional[str] = "", time_window: Optional[timedelta] = None, + sliding_window: Optional[timedelta] = None, ): self.column = column or "" self.function = function or "" self.time_window = time_window + if not self.sliding_window: + self.sliding_window = self.time_window + else: + self.sliding_window = self.time_window def to_proto(self) -> AggregationProto: window_duration = None @@ -36,8 +43,13 @@ def to_proto(self) -> AggregationProto: window_duration = Duration() window_duration.FromTimedelta(self.time_window) + sliding_window_duration = None + if self.sliding_window is not None: + sliding_window_duration = Duration() + sliding_window_duration.FromTimedelta(self.sliding_window) + return AggregationProto( - column=self.column, function=self.function, time_window=window_duration + column=self.column, function=self.function, time_window=window_duration, sliding_window=sliding_window_duration, ) @classmethod @@ -48,10 +60,17 @@ def from_proto(cls, agg_proto: AggregationProto): else agg_proto.time_window.ToTimedelta() ) + sliding_window = ( + timedelta(days=0) + if agg_proto.sliding_window.ToNanoseconds() == 0 + else agg_proto.sliding_window.ToTimedelta() + ) aggregation = cls( column=agg_proto.column, function=agg_proto.function, time_window=time_window, + sliding_window=sliding_window, + ) return aggregation @@ -63,6 +82,7 @@ def __eq__(self, other): self.column != other.column or self.function != other.function or self.time_window != other.time_window + or self.sliding_window != other.sliding_window ): return False From 94c6f0dd132656364bbcb3516b6ca54a3922a160 Mon Sep 17 00:00:00 2001 From: Kevin Zhang Date: Mon, 6 Jun 2022 14:43:38 -0700 Subject: [PATCH 2/7] Fix Signed-off-by: Kevin Zhang --- sdk/python/feast/aggregation.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/python/feast/aggregation.py b/sdk/python/feast/aggregation.py index d484f876aa3..af8e148f13d 100644 --- a/sdk/python/feast/aggregation.py +++ b/sdk/python/feast/aggregation.py @@ -35,7 +35,7 @@ def __init__( if not self.sliding_window: self.sliding_window = self.time_window else: - self.sliding_window = self.time_window + self.sliding_window = sliding_window def to_proto(self) -> AggregationProto: window_duration = None From 8f8cd7b9331e4fe14c64e565195ad048732d06cf Mon Sep 17 00:00:00 2001 From: Kevin Zhang Date: Tue, 7 Jun 2022 12:06:57 -0700 Subject: [PATCH 3/7] update apis Signed-off-by: Kevin Zhang --- protos/feast/core/DataSource.proto | 2 + sdk/python/feast/aggregation.py | 2 +- sdk/python/feast/data_source.py | 57 ++++++++++++++++++- .../integration/registration/test_registry.py | 3 + .../test_stream_feature_view_apply.py | 1 + 5 files changed, 62 insertions(+), 3 deletions(-) diff --git a/protos/feast/core/DataSource.proto b/protos/feast/core/DataSource.proto index 9e6028ccfa4..e71066ee709 100644 --- a/protos/feast/core/DataSource.proto +++ b/protos/feast/core/DataSource.proto @@ -22,6 +22,7 @@ option go_package = "github.com/feast-dev/feast/go/protos/feast/core"; option java_outer_classname = "DataSourceProto"; option java_package = "feast.proto.core"; +import "google/protobuf/duration.proto"; import "feast/core/DataFormat.proto"; import "feast/types/Value.proto"; import "feast/core/Feature.proto"; @@ -135,6 +136,7 @@ message DataSource { // Defines the stream data format encoding feature/entity data in Kafka messages. StreamFormat message_format = 3; + google.protobuf.Duration watermark = 4; } // Defines options for DataSource that sources features from Kinesis records. diff --git a/sdk/python/feast/aggregation.py b/sdk/python/feast/aggregation.py index af8e148f13d..c92fea0b0c1 100644 --- a/sdk/python/feast/aggregation.py +++ b/sdk/python/feast/aggregation.py @@ -32,7 +32,7 @@ def __init__( self.column = column or "" self.function = function or "" self.time_window = time_window - if not self.sliding_window: + if not sliding_window: self.sliding_window = self.time_window else: self.sliding_window = sliding_window diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py index 6f416e70d3d..f9b3b613e33 100644 --- a/sdk/python/feast/data_source.py +++ b/sdk/python/feast/data_source.py @@ -16,8 +16,10 @@ import warnings from abc import ABC, abstractmethod from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union +from datetime import timedelta from google.protobuf.json_format import MessageToJson +from google.protobuf.duration_pb2 import Duration from feast import type_map from feast.data_format import StreamFormat @@ -47,11 +49,12 @@ class KafkaOptions: """ def __init__( - self, bootstrap_servers: str, message_format: StreamFormat, topic: str, + self, bootstrap_servers: str, message_format: StreamFormat, topic: str, watermark: Optional[timedelta] = None ): self.bootstrap_servers = bootstrap_servers self.message_format = message_format self.topic = topic + self.watermark = watermark or None @classmethod def from_proto(cls, kafka_options_proto: DataSourceProto.KafkaOptions): @@ -64,11 +67,18 @@ def from_proto(cls, kafka_options_proto: DataSourceProto.KafkaOptions): Returns: Returns a BigQueryOptions object based on the kafka_options protobuf """ - + watermark = None + if kafka_options_proto.HasField("watermark"): + watermark=( + timedelta(days=0) + if kafka_options_proto.watermark.ToNanoseconds() == 0 + else kafka_options_proto.watermark.ToTimedelta() + ) kafka_options = cls( bootstrap_servers=kafka_options_proto.bootstrap_servers, message_format=StreamFormat.from_proto(kafka_options_proto.message_format), topic=kafka_options_proto.topic, + watermark=watermark, ) return kafka_options @@ -80,11 +90,18 @@ def to_proto(self) -> DataSourceProto.KafkaOptions: Returns: KafkaOptionsProto protobuf """ + watermark_duration = None + if self.watermark is not None: + watermark_duration = Duration() + print("ASdfasdf") + print(self.watermark) + watermark_duration.FromTimedelta(self.watermark) kafka_options_proto = DataSourceProto.KafkaOptions( bootstrap_servers=self.bootstrap_servers, message_format=self.message_format.to_proto(), topic=self.topic, + watermark=watermark_duration, ) return kafka_options_proto @@ -369,7 +386,32 @@ def __init__( owner: Optional[str] = "", timestamp_field: Optional[str] = "", batch_source: Optional[DataSource] = None, + watermark: Optional[timedelta] = None, ): + """ + Creates a KafkaSource stream source object. + Args: + name: str. Name of data source, which should be unique within a project + event_timestamp_column (optional): str. (Deprecated) Event timestamp column used for point in time + joins of feature values. + bootstrap_servers: str. The servers of the kafka broker in the form "localhost:9092". + message_format: StreamFormat. StreamFormat of serialized messages. + topic: str. The name of the topic to read from in the kafka source. + created_timestamp_column (optional): str. Timestamp column indicating when the row + was created, used for deduplicating rows. + field_mapping (optional): dict(str, str). A dictionary mapping of column names in this data + source to feature names in a feature table or view. Only used for feature + columns, not entity or timestamp columns. + date_partition_column (optional): str. Timestamp column used for partitioning. + description (optional): str. A human-readable description. + tags (optional): dict(str, str). A dictionary of key-value pairs to store arbitrary metadata. + owner (optional): str. The owner of the data source, typically the email of the primary + maintainer. + timestamp_field (optional): str. Event timestamp field used for point + in time joins of feature values. + batch_source: DataSource. The datasource that acts as a batch source. + watermark: timedelta. The watermark for stream data. Specifically how late stream data can arrive without being discarded. + """ positional_attributes = [ "name", "event_timestamp_column", @@ -425,10 +467,12 @@ def __init__( timestamp_field=timestamp_field, ) self.batch_source = batch_source + self.kafka_options = KafkaOptions( bootstrap_servers=_bootstrap_servers, message_format=_message_format, topic=_topic, + watermark=watermark, ) def __eq__(self, other): @@ -445,6 +489,7 @@ def __eq__(self, other): != other.kafka_options.bootstrap_servers or self.kafka_options.message_format != other.kafka_options.message_format or self.kafka_options.topic != other.kafka_options.topic + or self.kafka_options.watermark != other.kafka_options.watermark ): return False @@ -455,6 +500,13 @@ def __hash__(self): @staticmethod def from_proto(data_source: DataSourceProto): + watermark = None + if data_source.kafka_options.HasField("watermark"): + watermark=( + timedelta(days=0) + if data_source.kafka_options.watermark.ToNanoseconds() == 0 + else data_source.kafka_options.watermark.ToTimedelta() + ) return KafkaSource( name=data_source.name, event_timestamp_column=data_source.timestamp_field, @@ -463,6 +515,7 @@ def from_proto(data_source: DataSourceProto): message_format=StreamFormat.from_proto( data_source.kafka_options.message_format ), + watermark=watermark, topic=data_source.kafka_options.topic, created_timestamp_column=data_source.created_timestamp_column, timestamp_field=data_source.timestamp_field, diff --git a/sdk/python/tests/integration/registration/test_registry.py b/sdk/python/tests/integration/registration/test_registry.py index 222eb116d26..60c0d4aaabb 100644 --- a/sdk/python/tests/integration/registration/test_registry.py +++ b/sdk/python/tests/integration/registration/test_registry.py @@ -14,6 +14,7 @@ import time from datetime import timedelta from tempfile import mkstemp +from isort import stream import pandas as pd import pytest @@ -319,6 +320,7 @@ def simple_udf(x: int): message_format=AvroFormat(""), topic="topic", batch_source=FileSource(path="some path"), + watermark=timedelta(days=1), ) sfv = StreamFeatureView( @@ -353,6 +355,7 @@ def simple_udf(x: int): # List Feature Views assert len(stream_feature_views) == 1 + print(stream_feature_views[0]) assert stream_feature_views[0] == sfv test_registry.delete_feature_view("test kafka stream feature view", project) diff --git a/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py b/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py index d24618b2704..e19641f291e 100644 --- a/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py +++ b/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py @@ -27,6 +27,7 @@ def test_apply_stream_feature_view(environment) -> None: message_format=AvroFormat(""), topic="topic", batch_source=FileSource(path="test_path", timestamp_field="event_timestamp"), + watermark=timedelta(days=1), ) @stream_feature_view( From 7c010932eae667707c44a27362c0f2263816bb6a Mon Sep 17 00:00:00 2001 From: Kevin Zhang Date: Tue, 7 Jun 2022 12:08:15 -0700 Subject: [PATCH 4/7] Lint Signed-off-by: Kevin Zhang --- sdk/python/feast/aggregation.py | 6 ++++-- sdk/python/feast/data_source.py | 14 +++++++++----- .../integration/registration/test_registry.py | 2 +- 3 files changed, 14 insertions(+), 8 deletions(-) diff --git a/sdk/python/feast/aggregation.py b/sdk/python/feast/aggregation.py index c92fea0b0c1..ab1d60c43e7 100644 --- a/sdk/python/feast/aggregation.py +++ b/sdk/python/feast/aggregation.py @@ -49,7 +49,10 @@ def to_proto(self) -> AggregationProto: sliding_window_duration.FromTimedelta(self.sliding_window) return AggregationProto( - column=self.column, function=self.function, time_window=window_duration, sliding_window=sliding_window_duration, + column=self.column, + function=self.function, + time_window=window_duration, + sliding_window=sliding_window_duration, ) @classmethod @@ -70,7 +73,6 @@ def from_proto(cls, agg_proto: AggregationProto): function=agg_proto.function, time_window=time_window, sliding_window=sliding_window, - ) return aggregation diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py index f9b3b613e33..f88fabc21c5 100644 --- a/sdk/python/feast/data_source.py +++ b/sdk/python/feast/data_source.py @@ -15,11 +15,11 @@ import enum import warnings from abc import ABC, abstractmethod -from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union from datetime import timedelta +from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union -from google.protobuf.json_format import MessageToJson from google.protobuf.duration_pb2 import Duration +from google.protobuf.json_format import MessageToJson from feast import type_map from feast.data_format import StreamFormat @@ -49,7 +49,11 @@ class KafkaOptions: """ def __init__( - self, bootstrap_servers: str, message_format: StreamFormat, topic: str, watermark: Optional[timedelta] = None + self, + bootstrap_servers: str, + message_format: StreamFormat, + topic: str, + watermark: Optional[timedelta] = None, ): self.bootstrap_servers = bootstrap_servers self.message_format = message_format @@ -69,7 +73,7 @@ def from_proto(cls, kafka_options_proto: DataSourceProto.KafkaOptions): """ watermark = None if kafka_options_proto.HasField("watermark"): - watermark=( + watermark = ( timedelta(days=0) if kafka_options_proto.watermark.ToNanoseconds() == 0 else kafka_options_proto.watermark.ToTimedelta() @@ -502,7 +506,7 @@ def __hash__(self): def from_proto(data_source: DataSourceProto): watermark = None if data_source.kafka_options.HasField("watermark"): - watermark=( + watermark = ( timedelta(days=0) if data_source.kafka_options.watermark.ToNanoseconds() == 0 else data_source.kafka_options.watermark.ToTimedelta() diff --git a/sdk/python/tests/integration/registration/test_registry.py b/sdk/python/tests/integration/registration/test_registry.py index 60c0d4aaabb..9dd820dc20f 100644 --- a/sdk/python/tests/integration/registration/test_registry.py +++ b/sdk/python/tests/integration/registration/test_registry.py @@ -14,10 +14,10 @@ import time from datetime import timedelta from tempfile import mkstemp -from isort import stream import pandas as pd import pytest +from isort import stream from pytest_lazyfixture import lazy_fixture from feast import FileSource From 88d141e5962021cd1c39192f02190b97f6dc244f Mon Sep 17 00:00:00 2001 From: Kevin Zhang Date: Tue, 7 Jun 2022 12:22:04 -0700 Subject: [PATCH 5/7] Fix lint Signed-off-by: Kevin Zhang --- sdk/python/feast/aggregation.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/python/feast/aggregation.py b/sdk/python/feast/aggregation.py index ab1d60c43e7..dd65c1ab5cc 100644 --- a/sdk/python/feast/aggregation.py +++ b/sdk/python/feast/aggregation.py @@ -20,7 +20,7 @@ class Aggregation: column: str function: str time_window: Optional[timedelta] - sliding_window: timedelta + sliding_window: Optional[timedelta] def __init__( self, From 6ae89af96b28aee53370027023b94abfd6bbd908 Mon Sep 17 00:00:00 2001 From: Kevin Zhang Date: Tue, 7 Jun 2022 12:23:53 -0700 Subject: [PATCH 6/7] Fix Signed-off-by: Kevin Zhang --- sdk/python/tests/integration/registration/test_registry.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/sdk/python/tests/integration/registration/test_registry.py b/sdk/python/tests/integration/registration/test_registry.py index 9dd820dc20f..fcf65570a06 100644 --- a/sdk/python/tests/integration/registration/test_registry.py +++ b/sdk/python/tests/integration/registration/test_registry.py @@ -17,7 +17,6 @@ import pandas as pd import pytest -from isort import stream from pytest_lazyfixture import lazy_fixture from feast import FileSource @@ -355,7 +354,6 @@ def simple_udf(x: int): # List Feature Views assert len(stream_feature_views) == 1 - print(stream_feature_views[0]) assert stream_feature_views[0] == sfv test_registry.delete_feature_view("test kafka stream feature view", project) From e98ac38d6aaa44f53704b790d9d73aea93a091d9 Mon Sep 17 00:00:00 2001 From: Kevin Zhang Date: Wed, 8 Jun 2022 10:55:25 -0700 Subject: [PATCH 7/7] Fix Signed-off-by: Kevin Zhang --- protos/feast/core/Aggregation.proto | 2 +- sdk/python/feast/aggregation.py | 32 ++++++++++++++--------------- sdk/python/feast/data_source.py | 2 -- 3 files changed, 17 insertions(+), 19 deletions(-) diff --git a/protos/feast/core/Aggregation.proto b/protos/feast/core/Aggregation.proto index c2cf4f8f193..d2d6cab7021 100644 --- a/protos/feast/core/Aggregation.proto +++ b/protos/feast/core/Aggregation.proto @@ -11,5 +11,5 @@ message Aggregation { string column = 1; string function = 2; google.protobuf.Duration time_window = 3; - google.protobuf.Duration sliding_window = 4; + google.protobuf.Duration slide_interval = 4; } \ No newline at end of file diff --git a/sdk/python/feast/aggregation.py b/sdk/python/feast/aggregation.py index dd65c1ab5cc..d0af49b4259 100644 --- a/sdk/python/feast/aggregation.py +++ b/sdk/python/feast/aggregation.py @@ -14,28 +14,28 @@ class Aggregation: column: str # Column name of the feature we are aggregating. function: str # Provided built in aggregations sum, max, min, count mean time_window: timedelta # The time window for this aggregation. - sliding_window: timedelta # The sliding window for these aggregations + slide_interval: timedelta # The sliding window for these aggregations """ column: str function: str time_window: Optional[timedelta] - sliding_window: Optional[timedelta] + slide_interval: Optional[timedelta] def __init__( self, column: Optional[str] = "", function: Optional[str] = "", time_window: Optional[timedelta] = None, - sliding_window: Optional[timedelta] = None, + slide_interval: Optional[timedelta] = None, ): self.column = column or "" self.function = function or "" self.time_window = time_window - if not sliding_window: - self.sliding_window = self.time_window + if not slide_interval: + self.slide_interval = self.time_window else: - self.sliding_window = sliding_window + self.slide_interval = slide_interval def to_proto(self) -> AggregationProto: window_duration = None @@ -43,16 +43,16 @@ def to_proto(self) -> AggregationProto: window_duration = Duration() window_duration.FromTimedelta(self.time_window) - sliding_window_duration = None - if self.sliding_window is not None: - sliding_window_duration = Duration() - sliding_window_duration.FromTimedelta(self.sliding_window) + slide_interval_duration = None + if self.slide_interval is not None: + slide_interval_duration = Duration() + slide_interval_duration.FromTimedelta(self.slide_interval) return AggregationProto( column=self.column, function=self.function, time_window=window_duration, - sliding_window=sliding_window_duration, + slide_interval=slide_interval_duration, ) @classmethod @@ -63,16 +63,16 @@ def from_proto(cls, agg_proto: AggregationProto): else agg_proto.time_window.ToTimedelta() ) - sliding_window = ( + slide_interval = ( timedelta(days=0) - if agg_proto.sliding_window.ToNanoseconds() == 0 - else agg_proto.sliding_window.ToTimedelta() + if agg_proto.slide_interval.ToNanoseconds() == 0 + else agg_proto.slide_interval.ToTimedelta() ) aggregation = cls( column=agg_proto.column, function=agg_proto.function, time_window=time_window, - sliding_window=sliding_window, + slide_interval=slide_interval, ) return aggregation @@ -84,7 +84,7 @@ def __eq__(self, other): self.column != other.column or self.function != other.function or self.time_window != other.time_window - or self.sliding_window != other.sliding_window + or self.slide_interval != other.slide_interval ): return False diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py index f88fabc21c5..1211edd54aa 100644 --- a/sdk/python/feast/data_source.py +++ b/sdk/python/feast/data_source.py @@ -97,8 +97,6 @@ def to_proto(self) -> DataSourceProto.KafkaOptions: watermark_duration = None if self.watermark is not None: watermark_duration = Duration() - print("ASdfasdf") - print(self.watermark) watermark_duration.FromTimedelta(self.watermark) kafka_options_proto = DataSourceProto.KafkaOptions(